Add tests and docs for first/last_over_time and rate (#130290)

This PR adds unit tests and docs for first_over_time, last_over_time, 
and rate. For the rate function, the tests currently only verify that
the output is a double, not the actual value.

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

@@ -0,0 +1,11 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Description**
+
+The earliest value of a field, where recency determined by the `@timestamp` field.
+
+::::{note}
+Available with the [TS](/reference/query-languages/esql/commands/source-commands.md#esql-ts) command in snapshot builds
+::::
+
+

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

@@ -0,0 +1,11 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Description**
+
+The latest value of a field, where recency determined by the `@timestamp` field.
+
+::::{note}
+Available with the [TS](/reference/query-languages/esql/commands/source-commands.md#esql-ts) command in snapshot builds
+::::
+
+

docs/reference/query-languages/esql/_snippets/functions/description/rate.md

@@ -0,0 +1,11 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Description**
+
+The rate of a counter field.
+
+::::{note}
+Available with the [TS](/reference/query-languages/esql/commands/source-commands.md#esql-ts) command in snapshot builds
+::::
+
+

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

@@ -0,0 +1,16 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Example**
+
+```esql
+TS k8s
+| STATS max_cost=max(first_over_time(network.cost)) BY cluster, time_bucket = bucket(@timestamp,1minute)
+```
+
+| max_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 |
+
+

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

@@ -0,0 +1,17 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Example**
+
+```esql
+TS k8s
+| STATS max_cost=max(last_over_time(network.cost)) BY cluster, time_bucket = bucket(@timestamp,1minute)
+```
+
+| max_cost:double | cluster:keyword | time_bucket:datetime |
+| --- | --- | --- |
+| 12.5 | staging | 2024-05-10T00:09:00.000Z |
+| 12.375 | prod | 2024-05-10T00:17:00.000Z |
+| 12.375 | qa | 2024-05-10T00:06:00.000Z |
+| 12.375 | qa | 2024-05-10T00:01:00.000Z |
+
+

docs/reference/query-languages/esql/_snippets/functions/examples/rate.md

@@ -0,0 +1,15 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Example**
+
+```esql
+TS k8s
+| STATS max(rate(network.total_bytes_in)) BY time_bucket = bucket(@timestamp,5minute)
+```
+
+| max(rate(network.total_bytes_in)): double | time_bucket:date |
+| --- | --- |
+| 6.980660660660663 | 2024-05-10T00:20:00.000Z |
+| 23.702205882352942 | 2024-05-10T00:15:00.000Z |
+
+

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

@@ -0,0 +1,26 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+## `FIRST_OVER_TIME` [esql-first_over_time]
+```{applies_to}
+stack: unavailable
+```
+
+**Syntax**
+
+:::{image} ../../../images/functions/first_over_time.svg
+:alt: Embedded
+:class: text-center
+:::
+
+
+:::{include} ../parameters/first_over_time.md
+:::
+
+:::{include} ../description/first_over_time.md
+:::
+
+:::{include} ../types/first_over_time.md
+:::
+
+:::{include} ../examples/first_over_time.md
+:::

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

@@ -0,0 +1,26 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+## `LAST_OVER_TIME` [esql-last_over_time]
+```{applies_to}
+stack: unavailable
+```
+
+**Syntax**
+
+:::{image} ../../../images/functions/last_over_time.svg
+:alt: Embedded
+:class: text-center
+:::
+
+
+:::{include} ../parameters/last_over_time.md
+:::
+
+:::{include} ../description/last_over_time.md
+:::
+
+:::{include} ../types/last_over_time.md
+:::
+
+:::{include} ../examples/last_over_time.md
+:::

docs/reference/query-languages/esql/_snippets/functions/layout/rate.md

@@ -0,0 +1,26 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+## `RATE` [esql-rate]
+```{applies_to}
+stack: unavailable
+```
+
+**Syntax**
+
+:::{image} ../../../images/functions/rate.svg
+:alt: Embedded
+:class: text-center
+:::
+
+
+:::{include} ../parameters/rate.md
+:::
+
+:::{include} ../description/rate.md
+:::
+
+:::{include} ../types/rate.md
+:::
+
+:::{include} ../examples/rate.md
+:::

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

@@ -0,0 +1,7 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Parameters**
+
+`field`
+:   
+

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

@@ -0,0 +1,7 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Parameters**
+
+`field`
+:   
+

docs/reference/query-languages/esql/_snippets/functions/parameters/rate.md

@@ -0,0 +1,7 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Parameters**
+
+`field`
+:   
+

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

@@ -0,0 +1,10 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Supported types**
+
+| field | result |
+| --- | --- |
+| double | double |
+| integer | integer |
+| long | long |
+

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

@@ -0,0 +1,10 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Supported types**
+
+| field | result |
+| --- | --- |
+| double | double |
+| integer | integer |
+| long | long |
+

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

@@ -0,0 +1,10 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Supported types**
+
+| field | result |
+| --- | --- |
+| counter_double | double |
+| counter_integer | double |
+| counter_long | double |
+

docs/reference/query-languages/esql/_snippets/lists/aggregation-functions.md

@@ -16,3 +16,6 @@
 * [`TOP`](../../functions-operators/aggregation-functions.md#esql-top)
 * [preview] [`VALUES`](../../functions-operators/aggregation-functions.md#esql-values)
 * [`WEIGHTED_AVG`](../../functions-operators/aggregation-functions.md#esql-weighted_avg)
+* [unavailable] [`FIRST_OVER_TIME`](../../functions-operators/aggregation-functions.md#esql-first_over_time)
+* [unavailable] [`LAST_OVER_TIME`](../../functions-operators/aggregation-functions.md#esql-last_over_time)
+* [unavailable] [`RATE`](../../functions-operators/aggregation-functions.md#esql-rate)

docs/reference/query-languages/esql/functions-operators/aggregation-functions.md

@@ -66,3 +66,11 @@ The [`STATS`](/reference/query-languages/esql/commands/processing-commands.md#es
 :::{include} ../_snippets/functions/layout/weighted_avg.md
 :::
 
+:::{include} ../_snippets/functions/layout/first_over_time.md
+:::
+
+:::{include} ../_snippets/functions/layout/last_over_time.md
+:::
+
+:::{include} ../_snippets/functions/layout/rate.md
+:::

docs/reference/query-languages/esql/images/functions/first_over_time.svg

@@ -0,0 +1 @@
+<svg version="1.1" xmlns:xlink="http://www.w3.org/1999/xlink" xmlns="http://www.w3.org/2000/svg" width="384" height="46" viewbox="0 0 384 46"><defs><style type="text/css">.c{fill:none;stroke:#222222;}.k{fill:#000000;font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;font-size:20px;}.s{fill:#e4f4ff;stroke:#222222;}.syn{fill:#8D8D8D;font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;font-size:20px;}</style></defs><path class="c" d="M0 31h5m200 0h10m32 0h10m80 0h10m32 0h5"/><rect class="s" x="5" y="5" width="200" height="36"/><text class="k" x="15" y="31">FIRST_OVER_TIME</text><rect class="s" x="215" y="5" width="32" height="36" rx="7"/><text class="syn" x="225" y="31">(</text><rect class="s" x="257" y="5" width="80" height="36" rx="7"/><text class="k" x="267" y="31">field</text><rect class="s" x="347" y="5" width="32" height="36" rx="7"/><text class="syn" x="357" y="31">)</text></svg>

docs/reference/query-languages/esql/images/functions/last_over_time.svg

@@ -0,0 +1 @@
+<svg version="1.1" xmlns:xlink="http://www.w3.org/1999/xlink" xmlns="http://www.w3.org/2000/svg" width="372" height="46" viewbox="0 0 372 46"><defs><style type="text/css">.c{fill:none;stroke:#222222;}.k{fill:#000000;font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;font-size:20px;}.s{fill:#e4f4ff;stroke:#222222;}.syn{fill:#8D8D8D;font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;font-size:20px;}</style></defs><path class="c" d="M0 31h5m188 0h10m32 0h10m80 0h10m32 0h5"/><rect class="s" x="5" y="5" width="188" height="36"/><text class="k" x="15" y="31">LAST_OVER_TIME</text><rect class="s" x="203" y="5" width="32" height="36" rx="7"/><text class="syn" x="213" y="31">(</text><rect class="s" x="245" y="5" width="80" height="36" rx="7"/><text class="k" x="255" y="31">field</text><rect class="s" x="335" y="5" width="32" height="36" rx="7"/><text class="syn" x="345" y="31">)</text></svg>

docs/reference/query-languages/esql/images/functions/rate.svg

@@ -0,0 +1 @@
+<svg version="1.1" xmlns:xlink="http://www.w3.org/1999/xlink" xmlns="http://www.w3.org/2000/svg" width="252" height="46" viewbox="0 0 252 46"><defs><style type="text/css">.c{fill:none;stroke:#222222;}.k{fill:#000000;font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;font-size:20px;}.s{fill:#e4f4ff;stroke:#222222;}.syn{fill:#8D8D8D;font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;font-size:20px;}</style></defs><path class="c" d="M0 31h5m68 0h10m32 0h10m80 0h10m32 0h5"/><rect class="s" x="5" y="5" width="68" height="36"/><text class="k" x="15" y="31">RATE</text><rect class="s" x="83" y="5" width="32" height="36" rx="7"/><text class="syn" x="93" y="31">(</text><rect class="s" x="125" y="5" width="80" height="36" rx="7"/><text class="k" x="135" y="31">field</text><rect class="s" x="215" y="5" width="32" height="36" rx="7"/><text class="syn" x="225" y="31">)</text></svg>

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

@@ -0,0 +1,50 @@
+{
+  "comment" : "This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.",
+  "type" : "time_series_agg",
+  "name" : "first_over_time",
+  "description" : "The earliest value of a field, where recency determined by the `@timestamp` field.",
+  "note" : "Available with the TS command in snapshot builds",
+  "signatures" : [
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "double",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "double"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "integer",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "integer"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "long",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "long"
+    }
+  ],
+  "examples" : [
+    "TS k8s\n| STATS max_cost=max(first_over_time(network.cost)) BY cluster, time_bucket = bucket(@timestamp,1minute)"
+  ],
+  "preview" : false,
+  "snapshot_only" : true
+}

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

@@ -0,0 +1,50 @@
+{
+  "comment" : "This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.",
+  "type" : "time_series_agg",
+  "name" : "last_over_time",
+  "description" : "The latest value of a field, where recency determined by the `@timestamp` field.",
+  "note" : "Available with the TS command in snapshot builds",
+  "signatures" : [
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "double",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "double"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "integer",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "integer"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "long",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "long"
+    }
+  ],
+  "examples" : [
+    "TS k8s\n| STATS max_cost=max(last_over_time(network.cost)) BY cluster, time_bucket = bucket(@timestamp,1minute)"
+  ],
+  "preview" : false,
+  "snapshot_only" : true
+}

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

@@ -0,0 +1,50 @@
+{
+  "comment" : "This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.",
+  "type" : "time_series_agg",
+  "name" : "rate",
+  "description" : "The rate of a counter field.",
+  "note" : "Available with the TS command in snapshot builds",
+  "signatures" : [
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "counter_double",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "double"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "counter_integer",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "double"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "counter_long",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "double"
+    }
+  ],
+  "examples" : [
+    "TS k8s\n| STATS max(rate(network.total_bytes_in)) BY time_bucket = bucket(@timestamp,5minute)"
+  ],
+  "preview" : false,
+  "snapshot_only" : true
+}

docs/reference/query-languages/esql/kibana/docs/functions/first_over_time.md

@@ -0,0 +1,10 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+### FIRST OVER TIME
+The earliest value of a field, where recency determined by the `@timestamp` field.
+
+```esql
+TS k8s
+| STATS max_cost=max(first_over_time(network.cost)) BY cluster, time_bucket = bucket(@timestamp,1minute)
+```
+Note: Available with the [TS](https://www.elastic.co/docs/reference/query-languages/esql/commands/source-commands#esql-ts) command in snapshot builds

docs/reference/query-languages/esql/kibana/docs/functions/last_over_time.md

@@ -0,0 +1,10 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+### LAST OVER TIME
+The latest value of a field, where recency determined by the `@timestamp` field.
+
+```esql
+TS k8s
+| STATS max_cost=max(last_over_time(network.cost)) BY cluster, time_bucket = bucket(@timestamp,1minute)
+```
+Note: Available with the [TS](https://www.elastic.co/docs/reference/query-languages/esql/commands/source-commands#esql-ts) command in snapshot builds

docs/reference/query-languages/esql/kibana/docs/functions/rate.md

@@ -0,0 +1,10 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+### RATE
+The rate of a counter field.
+
+```esql
+TS k8s
+| STATS max(rate(network.total_bytes_in)) BY time_bucket = bucket(@timestamp,5minute)
+```
+Note: Available with the [TS](https://www.elastic.co/docs/reference/query-languages/esql/commands/source-commands#esql-ts) command in snapshot builds

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

@@ -92,16 +92,20 @@ public class FirstOverTimeDoubleAggregator {
         }
 
         void collectValue(int groupId, long timestamp, double value) {
+            boolean updated = false;
             if (groupId < timestamps.size()) {
                 // TODO: handle multiple values?
                 if (groupId > maxGroupId || hasValue(groupId) == false || timestamps.get(groupId) > timestamp) {
                     timestamps.set(groupId, timestamp);
-                    values.set(groupId, value);
+                    updated = true;
                 }
             } else {
                 timestamps = bigArrays.grow(timestamps, groupId + 1);
-                values = bigArrays.grow(values, groupId + 1);
                 timestamps.set(groupId, timestamp);
+                updated = true;
+            }
+            if (updated) {
+                values = bigArrays.grow(values, groupId + 1);
                 values.set(groupId, value);
             }
             maxGroupId = Math.max(maxGroupId, groupId);

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

@@ -92,16 +92,20 @@ public class FirstOverTimeFloatAggregator {
         }
 
         void collectValue(int groupId, long timestamp, float value) {
+            boolean updated = false;
             if (groupId < timestamps.size()) {
                 // TODO: handle multiple values?
                 if (groupId > maxGroupId || hasValue(groupId) == false || timestamps.get(groupId) > timestamp) {
                     timestamps.set(groupId, timestamp);
-                    values.set(groupId, value);
+                    updated = true;
                 }
             } else {
                 timestamps = bigArrays.grow(timestamps, groupId + 1);
-                values = bigArrays.grow(values, groupId + 1);
                 timestamps.set(groupId, timestamp);
+                updated = true;
+            }
+            if (updated) {
+                values = bigArrays.grow(values, groupId + 1);
                 values.set(groupId, value);
             }
             maxGroupId = Math.max(maxGroupId, groupId);

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

@@ -92,16 +92,20 @@ public class FirstOverTimeIntAggregator {
         }
 
         void collectValue(int groupId, long timestamp, int value) {
+            boolean updated = false;
             if (groupId < timestamps.size()) {
                 // TODO: handle multiple values?
                 if (groupId > maxGroupId || hasValue(groupId) == false || timestamps.get(groupId) > timestamp) {
                     timestamps.set(groupId, timestamp);
-                    values.set(groupId, value);
+                    updated = true;
                 }
             } else {
                 timestamps = bigArrays.grow(timestamps, groupId + 1);
-                values = bigArrays.grow(values, groupId + 1);
                 timestamps.set(groupId, timestamp);
+                updated = true;
+            }
+            if (updated) {
+                values = bigArrays.grow(values, groupId + 1);
                 values.set(groupId, value);
             }
             maxGroupId = Math.max(maxGroupId, groupId);

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

@@ -90,16 +90,20 @@ public class FirstOverTimeLongAggregator {
         }
 
         void collectValue(int groupId, long timestamp, long value) {
+            boolean updated = false;
             if (groupId < timestamps.size()) {
                 // TODO: handle multiple values?
                 if (groupId > maxGroupId || hasValue(groupId) == false || timestamps.get(groupId) > timestamp) {
                     timestamps.set(groupId, timestamp);
-                    values.set(groupId, value);
+                    updated = true;
                 }
             } else {
                 timestamps = bigArrays.grow(timestamps, groupId + 1);
-                values = bigArrays.grow(values, groupId + 1);
                 timestamps.set(groupId, timestamp);
+                updated = true;
+            }
+            if (updated) {
+                values = bigArrays.grow(values, groupId + 1);
                 values.set(groupId, value);
             }
             maxGroupId = Math.max(maxGroupId, groupId);

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

@@ -92,16 +92,20 @@ public class LastOverTimeDoubleAggregator {
         }
 
         void collectValue(int groupId, long timestamp, double value) {
+            boolean updated = false;
             if (groupId < timestamps.size()) {
                 // TODO: handle multiple values?
                 if (groupId > maxGroupId || hasValue(groupId) == false || timestamps.get(groupId) < timestamp) {
                     timestamps.set(groupId, timestamp);
-                    values.set(groupId, value);
+                    updated = true;
                 }
             } else {
                 timestamps = bigArrays.grow(timestamps, groupId + 1);
-                values = bigArrays.grow(values, groupId + 1);
                 timestamps.set(groupId, timestamp);
+                updated = true;
+            }
+            if (updated) {
+                values = bigArrays.grow(values, groupId + 1);
                 values.set(groupId, value);
             }
             maxGroupId = Math.max(maxGroupId, groupId);

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

@@ -92,16 +92,20 @@ public class LastOverTimeFloatAggregator {
         }
 
         void collectValue(int groupId, long timestamp, float value) {
+            boolean updated = false;
             if (groupId < timestamps.size()) {
                 // TODO: handle multiple values?
                 if (groupId > maxGroupId || hasValue(groupId) == false || timestamps.get(groupId) < timestamp) {
                     timestamps.set(groupId, timestamp);
-                    values.set(groupId, value);
+                    updated = true;
                 }
             } else {
                 timestamps = bigArrays.grow(timestamps, groupId + 1);
-                values = bigArrays.grow(values, groupId + 1);
                 timestamps.set(groupId, timestamp);
+                updated = true;
+            }
+            if (updated) {
+                values = bigArrays.grow(values, groupId + 1);
                 values.set(groupId, value);
             }
             maxGroupId = Math.max(maxGroupId, groupId);

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

@@ -92,16 +92,20 @@ public class LastOverTimeIntAggregator {
         }
 
         void collectValue(int groupId, long timestamp, int value) {
+            boolean updated = false;
             if (groupId < timestamps.size()) {
                 // TODO: handle multiple values?
                 if (groupId > maxGroupId || hasValue(groupId) == false || timestamps.get(groupId) < timestamp) {
                     timestamps.set(groupId, timestamp);
-                    values.set(groupId, value);
+                    updated = true;
                 }
             } else {
                 timestamps = bigArrays.grow(timestamps, groupId + 1);
-                values = bigArrays.grow(values, groupId + 1);
                 timestamps.set(groupId, timestamp);
+                updated = true;
+            }
+            if (updated) {
+                values = bigArrays.grow(values, groupId + 1);
                 values.set(groupId, value);
             }
             maxGroupId = Math.max(maxGroupId, groupId);

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

@@ -90,16 +90,20 @@ public class LastOverTimeLongAggregator {
         }
 
         void collectValue(int groupId, long timestamp, long value) {
+            boolean updated = false;
             if (groupId < timestamps.size()) {
                 // TODO: handle multiple values?
                 if (groupId > maxGroupId || hasValue(groupId) == false || timestamps.get(groupId) < timestamp) {
                     timestamps.set(groupId, timestamp);
-                    values.set(groupId, value);
+                    updated = true;
                 }
             } else {
                 timestamps = bigArrays.grow(timestamps, groupId + 1);
-                values = bigArrays.grow(values, groupId + 1);
                 timestamps.set(groupId, timestamp);
+                updated = true;
+            }
+            if (updated) {
+                values = bigArrays.grow(values, groupId + 1);
                 values.set(groupId, value);
             }
             maxGroupId = Math.max(maxGroupId, groupId);

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/aggregation/X-ValueOverTimeAggregator.java.st

@@ -96,16 +96,20 @@ public class $Occurrence$OverTime$Type$Aggregator {
         }
 
         void collectValue(int groupId, long timestamp, $type$ value) {
+            boolean updated = false;
             if (groupId < timestamps.size()) {
                 // TODO: handle multiple values?
                 if (groupId > maxGroupId || hasValue(groupId) == false || timestamps.get(groupId) $if(Last)$<$else$>$endif$ timestamp) {
                     timestamps.set(groupId, timestamp);
-                    values.set(groupId, value);
+                    updated = true;
                 }
             } else {
                 timestamps = bigArrays.grow(timestamps, groupId + 1);
-                values = bigArrays.grow(values, groupId + 1);
                 timestamps.set(groupId, timestamp);
+                updated = true;
+            }
+            if (updated) {
+                values = bigArrays.grow(values, groupId + 1);
                 values.set(groupId, value);
             }
             maxGroupId = Math.max(maxGroupId, groupId);

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

@@ -83,11 +83,17 @@ bytes: double       | sum(rate(network.total_cost)): double | cluster: keyword
 
 oneRateWithBucket
 required_capability: metrics_command
-TS k8s | STATS max(rate(network.total_bytes_in)) BY time_bucket = bucket(@timestamp,5minute) | SORT time_bucket DESC | LIMIT 2;
+// tag::rate[]
+TS k8s
+| STATS max(rate(network.total_bytes_in)) BY time_bucket = bucket(@timestamp,5minute)
+// end::rate[]
+| SORT time_bucket DESC | LIMIT 2;
 
+// tag::rate-result[]
 max(rate(network.total_bytes_in)): double | time_bucket:date
 6.980660660660663                         | 2024-05-10T00:20:00.000Z
 23.702205882352942                        | 2024-05-10T00:15:00.000Z
+// end::rate-result[]
 ;
 
 twoRatesWithBucket
@@ -260,13 +266,19 @@ avg_cost:double    | cluster:keyword | time_bucket:datetime
 max_of_last_over_time
 required_capability: metrics_command
 required_capability: last_over_time
-TS k8s | STATS max_cost=max(last_over_time(network.cost)) BY cluster, time_bucket = bucket(@timestamp,1minute) | SORT max_cost DESC, time_bucket DESC, cluster | LIMIT 10;
+// tag::last_over_time[]
+TS k8s 
+| STATS max_cost=max(last_over_time(network.cost)) BY cluster, time_bucket = bucket(@timestamp,1minute) 
+// end::last_over_time[]
+| SORT max_cost DESC, time_bucket DESC, cluster | LIMIT 10;
 
+// tag::last_over_time-result[]
 max_cost:double | cluster:keyword | time_bucket:datetime    
 12.5            | staging         | 2024-05-10T00:09:00.000Z
 12.375          | prod            | 2024-05-10T00:17:00.000Z
 12.375          | qa              | 2024-05-10T00:06:00.000Z
 12.375          | qa              | 2024-05-10T00:01:00.000Z
+// end::last_over_time-result[]
 12.25           | prod            | 2024-05-10T00:19:00.000Z
 12.125          | qa              | 2024-05-10T00:17:00.000Z
 12.125          | prod            | 2024-05-10T00:00:00.000Z
@@ -278,12 +290,18 @@ max_cost:double | cluster:keyword | time_bucket:datetime
 max_of_first_over_time
 required_capability: metrics_command
 required_capability: first_over_time
-TS k8s | STATS max_cost=max(first_over_time(network.cost)) BY cluster, time_bucket = bucket(@timestamp,1minute) | SORT max_cost DESC, time_bucket DESC, cluster | LIMIT 10;
+// tag::first_over_time[]
+TS k8s
+| STATS max_cost=max(first_over_time(network.cost)) BY cluster, time_bucket = bucket(@timestamp,1minute)
+// end::first_over_time[]
+| SORT max_cost DESC, time_bucket DESC, cluster | LIMIT 10;
 
+// tag::first_over_time-result[]
 max_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
+// end::first_over_time-result[]
 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

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

@@ -468,15 +468,15 @@ public class EsqlFunctionRegistry {
                 // The delay() function is for debug/snapshot environments only and should never be enabled in a non-snapshot build.
                 // This is an experimental function and can be removed without notice.
                 def(Delay.class, Delay::new, "delay"),
-                def(Rate.class, Rate::withUnresolvedTimestamp, "rate"),
+                def(Rate.class, uni(Rate::new), "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(CountOverTime.class, uni(CountOverTime::new), "count_over_time"),
                 def(CountDistinctOverTime.class, bi(CountDistinctOverTime::new), "count_distinct_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"),
+                def(LastOverTime.class, uni(LastOverTime::new), "last_over_time"),
+                def(FirstOverTime.class, uni(FirstOverTime::new), "first_over_time"),
                 def(Term.class, bi(Term::new), "term"),
                 def(Knn.class, tri(Knn::new), "knn"),
                 def(StGeohash.class, StGeohash::new, "st_geohash"),

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

@@ -21,6 +21,9 @@ 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.Example;
+import org.elasticsearch.xpack.esql.expression.function.FunctionAppliesTo;
+import org.elasticsearch.xpack.esql.expression.function.FunctionAppliesToLifecycle;
 import org.elasticsearch.xpack.esql.expression.function.FunctionInfo;
 import org.elasticsearch.xpack.esql.expression.function.FunctionType;
 import org.elasticsearch.xpack.esql.expression.function.OptionalArgument;
@@ -32,6 +35,7 @@ import java.io.IOException;
 import java.util.List;
 
 import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.ParamOrdinal.DEFAULT;
+import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.ParamOrdinal.SECOND;
 import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isType;
 
 public class FirstOverTime extends TimeSeriesAggregateFunction implements OptionalArgument, ToAggregator {
@@ -43,16 +47,20 @@ public class FirstOverTime extends TimeSeriesAggregateFunction implements Option
 
     private final Expression timestamp;
 
+    // TODO: support all types
     @FunctionInfo(
-        returnType = { "int", "double", "integer", "long" },
-        description = "Collect the first occurrence value of a time-series in the specified interval. Available with TS command only",
-        type = FunctionType.AGGREGATE
+        type = FunctionType.TIME_SERIES_AGGREGATE,
+        returnType = { "long", "integer", "double" },
+        description = "The earliest value of a field, where recency determined by the `@timestamp` field.",
+        appliesTo = { @FunctionAppliesTo(lifeCycle = FunctionAppliesToLifecycle.UNAVAILABLE) },
+        note = "Available with the [TS](/reference/query-languages/esql/commands/source-commands.md#esql-ts) command in snapshot builds",
+        examples = { @Example(file = "k8s-timeseries", tag = "first_over_time") }
     )
-    public FirstOverTime(
-        Source source,
-        @Param(name = "field", type = { "long|int|double|float" }, description = "field") Expression field,
-        Expression timestamp
-    ) {
+    public FirstOverTime(Source source, @Param(name = "field", type = { "long", "integer", "double" }) Expression field) {
+        this(source, field, new UnresolvedAttribute(source, "@timestamp"));
+    }
+
+    public FirstOverTime(Source source, Expression field, Expression timestamp) {
         this(source, field, Literal.TRUE, timestamp);
     }
 
@@ -80,10 +88,6 @@ public class FirstOverTime extends TimeSeriesAggregateFunction implements Option
         return ENTRY.name;
     }
 
-    public static FirstOverTime withUnresolvedTimestamp(Source source, Expression field) {
-        return new FirstOverTime(source, field, new UnresolvedAttribute(source, "@timestamp"));
-    }
-
     @Override
     protected NodeInfo<FirstOverTime> info() {
         return NodeInfo.create(this, FirstOverTime::new, field(), timestamp);
@@ -110,7 +114,16 @@ public class FirstOverTime extends TimeSeriesAggregateFunction implements Option
 
     @Override
     protected TypeResolution resolveType() {
-        return isType(field(), dt -> dt.isNumeric() && dt != DataType.UNSIGNED_LONG, sourceText(), DEFAULT, "numeric except unsigned_long");
+        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"
+                )
+            );
     }
 
     @Override

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

@@ -21,6 +21,9 @@ 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.Example;
+import org.elasticsearch.xpack.esql.expression.function.FunctionAppliesTo;
+import org.elasticsearch.xpack.esql.expression.function.FunctionAppliesToLifecycle;
 import org.elasticsearch.xpack.esql.expression.function.FunctionInfo;
 import org.elasticsearch.xpack.esql.expression.function.FunctionType;
 import org.elasticsearch.xpack.esql.expression.function.OptionalArgument;
@@ -32,6 +35,7 @@ import java.io.IOException;
 import java.util.List;
 
 import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.ParamOrdinal.DEFAULT;
+import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.ParamOrdinal.SECOND;
 import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isType;
 
 public class LastOverTime extends TimeSeriesAggregateFunction implements OptionalArgument, ToAggregator {
@@ -43,16 +47,20 @@ public class LastOverTime extends TimeSeriesAggregateFunction implements Optiona
 
     private final Expression timestamp;
 
+    // TODO: support all types
     @FunctionInfo(
-        returnType = { "int", "double", "integer", "long" },
-        description = "Collect the most recent value of a time-series in the specified interval. Available with TS command only",
-        type = FunctionType.AGGREGATE
+        type = FunctionType.TIME_SERIES_AGGREGATE,
+        returnType = { "long", "integer", "double" },
+        description = "The latest value of a field, where recency determined by the `@timestamp` field.",
+        appliesTo = { @FunctionAppliesTo(lifeCycle = FunctionAppliesToLifecycle.UNAVAILABLE) },
+        note = "Available with the [TS](/reference/query-languages/esql/commands/source-commands.md#esql-ts) command in snapshot builds",
+        examples = { @Example(file = "k8s-timeseries", tag = "last_over_time") }
     )
-    public LastOverTime(
-        Source source,
-        @Param(name = "field", type = { "long|int|double|float" }, description = "field") Expression field,
-        Expression timestamp
-    ) {
+    public LastOverTime(Source source, @Param(name = "field", type = { "long", "integer", "double" }) Expression field) {
+        this(source, field, new UnresolvedAttribute(source, "@timestamp"));
+    }
+
+    public LastOverTime(Source source, Expression field, Expression timestamp) {
         this(source, field, Literal.TRUE, timestamp);
     }
 
@@ -80,10 +88,6 @@ public class LastOverTime extends TimeSeriesAggregateFunction implements Optiona
         return ENTRY.name;
     }
 
-    public static LastOverTime withUnresolvedTimestamp(Source source, Expression field) {
-        return new LastOverTime(source, field, new UnresolvedAttribute(source, "@timestamp"));
-    }
-
     @Override
     protected NodeInfo<LastOverTime> info() {
         return NodeInfo.create(this, LastOverTime::new, field(), timestamp);
@@ -110,7 +114,16 @@ public class LastOverTime extends TimeSeriesAggregateFunction implements Optiona
 
     @Override
     protected TypeResolution resolveType() {
-        return isType(field(), dt -> dt.isNumeric() && dt != DataType.UNSIGNED_LONG, sourceText(), DEFAULT, "numeric except unsigned_long");
+        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"
+                )
+            );
     }
 
     @Override

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

@@ -20,6 +20,9 @@ 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.Example;
+import org.elasticsearch.xpack.esql.expression.function.FunctionAppliesTo;
+import org.elasticsearch.xpack.esql.expression.function.FunctionAppliesToLifecycle;
 import org.elasticsearch.xpack.esql.expression.function.FunctionInfo;
 import org.elasticsearch.xpack.esql.expression.function.FunctionType;
 import org.elasticsearch.xpack.esql.expression.function.OptionalArgument;
@@ -39,13 +42,20 @@ public class Rate extends TimeSeriesAggregateFunction implements OptionalArgumen
     private final Expression timestamp;
 
     @FunctionInfo(
+        type = FunctionType.TIME_SERIES_AGGREGATE,
         returnType = { "double" },
-        description = "compute the rate of a counter field. Available in METRICS command only",
-        type = FunctionType.AGGREGATE
+        description = "The rate of a counter field.",
+        appliesTo = { @FunctionAppliesTo(lifeCycle = FunctionAppliesToLifecycle.UNAVAILABLE) },
+        note = "Available with the [TS](/reference/query-languages/esql/commands/source-commands.md#esql-ts) command in snapshot builds",
+        examples = { @Example(file = "k8s-timeseries", tag = "rate") }
     )
+    public Rate(Source source, @Param(name = "field", type = { "counter_long", "counter_integer", "counter_double" }) Expression field) {
+        this(source, field, new UnresolvedAttribute(source, "@timestamp"));
+    }
+
     public Rate(
         Source source,
-        @Param(name = "field", type = { "counter_long|counter_integer|counter_double" }, description = "counter field") Expression field,
+        @Param(name = "field", type = { "counter_long", "counter_integer", "counter_double" }) Expression field,
         Expression timestamp
     ) {
         this(source, field, Literal.TRUE, timestamp);
@@ -75,10 +85,6 @@ public class Rate extends TimeSeriesAggregateFunction implements OptionalArgumen
         return ENTRY.name;
     }
 
-    public static Rate withUnresolvedTimestamp(Source source, Expression field) {
-        return new Rate(source, field, new UnresolvedAttribute(source, "@timestamp"));
-    }
-
     @Override
     protected NodeInfo<Rate> info() {
         return NodeInfo.create(this, Rate::new, field(), timestamp);

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/AbstractFunctionTestCase.java

@@ -959,11 +959,24 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
             if (tc.getData().stream().anyMatch(t -> t.type() == DataType.NULL)) {
                 continue;
             }
-            signatures.putIfAbsent(tc.getData().stream().map(TestCaseSupplier.TypedData::type).toList(), tc.expectedType());
+            List<DataType> types = tc.getData().stream().map(TestCaseSupplier.TypedData::type).toList();
+            signatures.putIfAbsent(signatureTypes(testClass, types), tc.expectedType());
         }
         return signatures;
     }
 
+    @SuppressWarnings("unchecked")
+    private static List<DataType> signatureTypes(Class<?> testClass, List<DataType> types) {
+        try {
+            Method method = testClass.getMethod("signatureTypes", List.class);
+            return (List<DataType>) method.invoke(null, types);
+        } catch (NoSuchMethodException ingored) {
+            return types;
+        } catch (Exception e) {
+            throw new AssertionError(e);
+        }
+    }
+
     @AfterClass
     public static void renderDocs() throws Exception {
         if (System.getProperty("generateDocs") == null) {

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/FirstOverTimeTests.java

@@ -0,0 +1,99 @@
+/*
+ * 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 com.carrotsearch.randomizedtesting.annotations.Name;
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.expression.function.AbstractAggregationTestCase;
+import org.elasticsearch.xpack.esql.expression.function.MultiRowTestCaseSupplier;
+import org.elasticsearch.xpack.esql.expression.function.TestCaseSupplier;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Supplier;
+import java.util.stream.IntStream;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
+
+public class FirstOverTimeTests extends AbstractAggregationTestCase {
+    public FirstOverTimeTests(@Name("TestCase") Supplier<TestCaseSupplier.TestCase> testCaseSupplier) {
+        this.testCase = testCaseSupplier.get();
+    }
+
+    @ParametersFactory
+    public static Iterable<Object[]> parameters() {
+        var suppliers = new ArrayList<TestCaseSupplier>();
+
+        var valuesSuppliers = List.of(
+            MultiRowTestCaseSupplier.longCases(1, 1000, Long.MIN_VALUE, Long.MAX_VALUE, true),
+            MultiRowTestCaseSupplier.intCases(1, 1000, Integer.MIN_VALUE, Integer.MAX_VALUE, true),
+            MultiRowTestCaseSupplier.doubleCases(1, 1000, -Double.MAX_VALUE, Double.MAX_VALUE, true)
+
+        );
+        for (List<TestCaseSupplier.TypedDataSupplier> valuesSupplier : valuesSuppliers) {
+            for (TestCaseSupplier.TypedDataSupplier fieldSupplier : valuesSupplier) {
+                TestCaseSupplier testCaseSupplier = makeSupplier(fieldSupplier);
+                suppliers.add(testCaseSupplier);
+            }
+        }
+        return parameterSuppliersFromTypedDataWithDefaultChecksNoErrors(suppliers);
+    }
+
+    @Override
+    protected Expression build(Source source, List<Expression> args) {
+        return new FirstOverTime(source, args.get(0), args.get(1));
+    }
+
+    @Override
+    public void testAggregate() {
+        assumeTrue("time-series aggregation doesn't support ungrouped", false);
+    }
+
+    @Override
+    public void testAggregateIntermediate() {
+        assumeTrue("time-series aggregation doesn't support ungrouped", false);
+    }
+
+    private static TestCaseSupplier makeSupplier(TestCaseSupplier.TypedDataSupplier fieldSupplier) {
+        DataType type = fieldSupplier.type();
+        return new TestCaseSupplier(fieldSupplier.name(), List.of(type, DataType.DATETIME), () -> {
+            TestCaseSupplier.TypedData fieldTypedData = fieldSupplier.get();
+            List<Object> dataRows = fieldTypedData.multiRowData();
+            List<Long> timestamps = IntStream.range(0, dataRows.size()).mapToLong(unused -> randomNonNegativeLong()).boxed().toList();
+            TestCaseSupplier.TypedData timestampsField = TestCaseSupplier.TypedData.multiRow(timestamps, DataType.DATETIME, "timestamps");
+            Object expected = null;
+            long lastTimestamp = Long.MIN_VALUE;
+            for (int i = 0; i < dataRows.size(); i++) {
+                if (i == 0) {
+                    expected = dataRows.get(i);
+                    lastTimestamp = timestamps.get(i);
+                } else if (timestamps.get(i) < lastTimestamp) {
+                    expected = dataRows.get(i);
+                    lastTimestamp = timestamps.get(i);
+                }
+            }
+            return new TestCaseSupplier.TestCase(
+                List.of(fieldTypedData, timestampsField),
+                "FirstOverTime[field=Attribute[channel=0],timestamp=Attribute[channel=1]]",
+                type,
+                equalTo(expected)
+            );
+        });
+    }
+
+    public static List<DataType> signatureTypes(List<DataType> testCaseTypes) {
+        assertThat(testCaseTypes, hasSize(2));
+        assertThat(testCaseTypes.get(1), equalTo(DataType.DATETIME));
+        return List.of(testCaseTypes.get(0));
+    }
+}

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/LastOverTimeTests.java

@@ -0,0 +1,99 @@
+/*
+ * 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 com.carrotsearch.randomizedtesting.annotations.Name;
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.expression.function.AbstractAggregationTestCase;
+import org.elasticsearch.xpack.esql.expression.function.MultiRowTestCaseSupplier;
+import org.elasticsearch.xpack.esql.expression.function.TestCaseSupplier;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Supplier;
+import java.util.stream.IntStream;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
+
+public class LastOverTimeTests extends AbstractAggregationTestCase {
+    public LastOverTimeTests(@Name("TestCase") Supplier<TestCaseSupplier.TestCase> testCaseSupplier) {
+        this.testCase = testCaseSupplier.get();
+    }
+
+    @ParametersFactory
+    public static Iterable<Object[]> parameters() {
+        var suppliers = new ArrayList<TestCaseSupplier>();
+
+        var valuesSuppliers = List.of(
+            MultiRowTestCaseSupplier.longCases(1, 1000, Long.MIN_VALUE, Long.MAX_VALUE, true),
+            MultiRowTestCaseSupplier.intCases(1, 1000, Integer.MIN_VALUE, Integer.MAX_VALUE, true),
+            MultiRowTestCaseSupplier.doubleCases(1, 1000, -Double.MAX_VALUE, Double.MAX_VALUE, true)
+
+        );
+        for (List<TestCaseSupplier.TypedDataSupplier> valuesSupplier : valuesSuppliers) {
+            for (TestCaseSupplier.TypedDataSupplier fieldSupplier : valuesSupplier) {
+                TestCaseSupplier testCaseSupplier = makeSupplier(fieldSupplier);
+                suppliers.add(testCaseSupplier);
+            }
+        }
+        return parameterSuppliersFromTypedDataWithDefaultChecksNoErrors(suppliers);
+    }
+
+    @Override
+    protected Expression build(Source source, List<Expression> args) {
+        return new LastOverTime(source, args.get(0), args.get(1));
+    }
+
+    @Override
+    public void testAggregate() {
+        assumeTrue("time-series aggregation doesn't support ungrouped", false);
+    }
+
+    @Override
+    public void testAggregateIntermediate() {
+        assumeTrue("time-series aggregation doesn't support ungrouped", false);
+    }
+
+    private static TestCaseSupplier makeSupplier(TestCaseSupplier.TypedDataSupplier fieldSupplier) {
+        DataType type = fieldSupplier.type();
+        return new TestCaseSupplier(fieldSupplier.name(), List.of(type, DataType.DATETIME), () -> {
+            TestCaseSupplier.TypedData fieldTypedData = fieldSupplier.get();
+            List<Object> dataRows = fieldTypedData.multiRowData();
+            List<Long> timestamps = IntStream.range(0, dataRows.size()).mapToLong(unused -> randomNonNegativeLong()).boxed().toList();
+            TestCaseSupplier.TypedData timestampsField = TestCaseSupplier.TypedData.multiRow(timestamps, DataType.DATETIME, "timestamps");
+            Object expected = null;
+            long lastTimestamp = Long.MIN_VALUE;
+            for (int i = 0; i < dataRows.size(); i++) {
+                if (i == 0) {
+                    expected = dataRows.get(i);
+                    lastTimestamp = timestamps.get(i);
+                } else if (timestamps.get(i) > lastTimestamp) {
+                    expected = dataRows.get(i);
+                    lastTimestamp = timestamps.get(i);
+                }
+            }
+            return new TestCaseSupplier.TestCase(
+                List.of(fieldTypedData, timestampsField),
+                "LastOverTime[field=Attribute[channel=0],timestamp=Attribute[channel=1]]",
+                type,
+                equalTo(expected)
+            );
+        });
+    }
+
+    public static List<DataType> signatureTypes(List<DataType> testCaseTypes) {
+        assertThat(testCaseTypes, hasSize(2));
+        assertThat(testCaseTypes.get(1), equalTo(DataType.DATETIME));
+        return List.of(testCaseTypes.get(0));
+    }
+}

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/RateTests.java

@@ -0,0 +1,115 @@
+/*
+ * 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 com.carrotsearch.randomizedtesting.annotations.Name;
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.expression.function.AbstractAggregationTestCase;
+import org.elasticsearch.xpack.esql.expression.function.MultiRowTestCaseSupplier;
+import org.elasticsearch.xpack.esql.expression.function.TestCaseSupplier;
+import org.hamcrest.Matcher;
+import org.hamcrest.Matchers;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Supplier;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
+
+public class RateTests extends AbstractAggregationTestCase {
+    public RateTests(@Name("TestCase") Supplier<TestCaseSupplier.TestCase> testCaseSupplier) {
+        this.testCase = testCaseSupplier.get();
+    }
+
+    @ParametersFactory
+    public static Iterable<Object[]> parameters() {
+        var suppliers = new ArrayList<TestCaseSupplier>();
+
+        var valuesSuppliers = List.of(
+            MultiRowTestCaseSupplier.longCases(1, 1000, 0, 1000_000_000, true),
+            MultiRowTestCaseSupplier.intCases(1, 1000, 0, 1000_000_000, true),
+            MultiRowTestCaseSupplier.doubleCases(1, 1000, 0, 1000_000_000, true)
+
+        );
+        for (List<TestCaseSupplier.TypedDataSupplier> valuesSupplier : valuesSuppliers) {
+            for (TestCaseSupplier.TypedDataSupplier fieldSupplier : valuesSupplier) {
+                TestCaseSupplier testCaseSupplier = makeSupplier(fieldSupplier);
+                suppliers.add(testCaseSupplier);
+            }
+        }
+        return parameterSuppliersFromTypedDataWithDefaultChecksNoErrors(suppliers);
+    }
+
+    @Override
+    protected Expression build(Source source, List<Expression> args) {
+        return new Rate(source, args.get(0), args.get(1));
+    }
+
+    @Override
+    public void testAggregate() {
+        assumeTrue("time-series aggregation doesn't support ungrouped", false);
+    }
+
+    @Override
+    public void testAggregateIntermediate() {
+        assumeTrue("time-series aggregation doesn't support ungrouped", false);
+    }
+
+    private static DataType counterType(DataType type) {
+        return switch (type) {
+            case DOUBLE -> DataType.COUNTER_DOUBLE;
+            case LONG -> DataType.COUNTER_LONG;
+            case INTEGER -> DataType.COUNTER_INTEGER;
+            default -> throw new AssertionError("unknown type for counter: " + type);
+        };
+    }
+
+    private static TestCaseSupplier makeSupplier(TestCaseSupplier.TypedDataSupplier fieldSupplier) {
+        DataType type = counterType(fieldSupplier.type());
+        return new TestCaseSupplier(fieldSupplier.name(), List.of(type, DataType.DATETIME), () -> {
+            TestCaseSupplier.TypedData fieldTypedData = fieldSupplier.get();
+            List<Object> dataRows = fieldTypedData.multiRowData();
+            fieldTypedData = TestCaseSupplier.TypedData.multiRow(dataRows, type, fieldTypedData.name());
+            List<Long> timestamps = new ArrayList<>();
+            long lastTimestamp = randomLongBetween(0, 1_000_000);
+            for (int row = 0; row < dataRows.size(); row++) {
+                lastTimestamp += randomLongBetween(1, 10_000);
+                timestamps.add(lastTimestamp);
+            }
+            TestCaseSupplier.TypedData timestampsField = TestCaseSupplier.TypedData.multiRow(
+                timestamps.reversed(),
+                DataType.DATETIME,
+                "timestamps"
+            );
+            final Matcher<?> matcher;
+            if (dataRows.size() < 2) {
+                matcher = Matchers.nullValue();
+            } else {
+                // TODO: check the value?
+                matcher = Matchers.allOf(Matchers.greaterThanOrEqualTo(0.0), Matchers.lessThan(Double.POSITIVE_INFINITY));
+            }
+            return new TestCaseSupplier.TestCase(
+                List.of(fieldTypedData, timestampsField),
+                "Rate[field=Attribute[channel=0],timestamp=Attribute[channel=1]]",
+                DataType.DOUBLE,
+                matcher
+            );
+        });
+    }
+
+    public static List<DataType> signatureTypes(List<DataType> testCaseTypes) {
+        assertThat(testCaseTypes, hasSize(2));
+        assertThat(testCaseTypes.get(1), equalTo(DataType.DATETIME));
+        return List.of(testCaseTypes.get(0));
+    }
+}