ESQL: Prototype FIRST/LAST (#132603)

Prototype implementation of FIRST and LAST. FIRST returns the value with the earliest timestamp. LAST returns the values with the latest timestamp.

Looks like:
```
FROM k8s
| STATS last_bytes_in = LAST(network.bytes_in, @timestamp) BY pod
| SORT pod ASC
```

SNAPSHOT only for now while we resolve the remaining open questions in #108385.

Important: FIRST/LAST in this PR only return a single value. In the example above, if the value of `network.bytes_in` with the latest `@timestamp` is `null` then we get the value with the next highest timestamp. If the value of `network.bytes_in` with the latest `@timestamp` is multivalued we'll get a random value from the top values. Some folks want that behavior, but surely not everyone. We'll figure out what it should do soon. But we can get this in under snapshot and folks can play with it.

In this prototype, if two documents tie in `@timestamp` then you'll get a value from one of them. Which one is undefined.

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

@@ -0,0 +1,6 @@
+% 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.
+

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

@@ -0,0 +1,6 @@
+% 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.
+

docs/reference/query-languages/esql/_snippets/functions/examples/first.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
+FROM k8s
+| STATS first_bytes_in = FIRST(network.bytes_in, @timestamp) BY pod
+| SORT pod ASC
+```
+
+| first_bytes_in:long | pod:keyword |
+| --- | --- |
+| 278 | one |
+| 473 | three |
+| 699 | two |
+
+

docs/reference/query-languages/esql/_snippets/functions/examples/last.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
+FROM k8s
+| STATS last_bytes_in = LAST(network.bytes_in, @timestamp) BY pod
+| SORT pod ASC
+```
+
+| last_bytes_in:long | pod:keyword |
+| --- | --- |
+| 206 | one |
+| 972 | three |
+| 812 | two |
+
+

docs/reference/query-languages/esql/_snippets/functions/layout/first.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` [esql-first]
+```{applies_to}
+stack: unavailable
+```
+
+**Syntax**
+
+:::{image} ../../../images/functions/first.svg
+:alt: Embedded
+:class: text-center
+:::
+
+
+:::{include} ../parameters/first.md
+:::
+
+:::{include} ../description/first.md
+:::
+
+:::{include} ../types/first.md
+:::
+
+:::{include} ../examples/first.md
+:::

docs/reference/query-languages/esql/_snippets/functions/layout/last.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` [esql-last]
+```{applies_to}
+stack: unavailable
+```
+
+**Syntax**
+
+:::{image} ../../../images/functions/last.svg
+:alt: Embedded
+:class: text-center
+:::
+
+
+:::{include} ../parameters/last.md
+:::
+
+:::{include} ../description/last.md
+:::
+
+:::{include} ../types/last.md
+:::
+
+:::{include} ../examples/last.md
+:::

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

@@ -0,0 +1,10 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Parameters**
+
+`value`
+:   Values to return
+
+`sort`
+:   Sort key
+

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

@@ -0,0 +1,10 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Parameters**
+
+`value`
+:   Values to return
+
+`sort`
+:   Sort key
+

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

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

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

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

docs/reference/query-languages/esql/images/functions/first.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 31h5m80 0h10m32 0h10m80 0h10m32 0h10m68 0h10m32 0h5"/><rect class="s" x="5" y="5" width="80" height="36"/><text class="k" x="15" y="31">FIRST</text><rect class="s" x="95" y="5" width="32" height="36" rx="7"/><text class="syn" x="105" y="31">(</text><rect class="s" x="137" y="5" width="80" height="36" rx="7"/><text class="k" x="147" y="31">value</text><rect class="s" x="227" y="5" width="32" height="36" rx="7"/><text class="syn" x="237" y="31">,</text><rect class="s" x="269" y="5" width="68" height="36" rx="7"/><text class="k" x="279" y="31">sort</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.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 31h5m68 0h10m32 0h10m80 0h10m32 0h10m68 0h10m32 0h5"/><rect class="s" x="5" y="5" width="68" height="36"/><text class="k" x="15" y="31">LAST</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">value</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="68" height="36" rx="7"/><text class="k" x="267" y="31">sort</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/kibana/definition/functions/first.json

@@ -0,0 +1,121 @@
+{
+  "comment" : "This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.",
+  "type" : "agg",
+  "name" : "first",
+  "description" : "The earliest value of a field.",
+  "signatures" : [
+    {
+      "params" : [
+        {
+          "name" : "value",
+          "type" : "double",
+          "optional" : false,
+          "description" : "Values to return"
+        },
+        {
+          "name" : "sort",
+          "type" : "date",
+          "optional" : false,
+          "description" : "Sort key"
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "double"
+    },
+    {
+      "params" : [
+        {
+          "name" : "value",
+          "type" : "double",
+          "optional" : false,
+          "description" : "Values to return"
+        },
+        {
+          "name" : "sort",
+          "type" : "date_nanos",
+          "optional" : false,
+          "description" : "Sort key"
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "double"
+    },
+    {
+      "params" : [
+        {
+          "name" : "value",
+          "type" : "integer",
+          "optional" : false,
+          "description" : "Values to return"
+        },
+        {
+          "name" : "sort",
+          "type" : "date",
+          "optional" : false,
+          "description" : "Sort key"
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "integer"
+    },
+    {
+      "params" : [
+        {
+          "name" : "value",
+          "type" : "integer",
+          "optional" : false,
+          "description" : "Values to return"
+        },
+        {
+          "name" : "sort",
+          "type" : "date_nanos",
+          "optional" : false,
+          "description" : "Sort key"
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "integer"
+    },
+    {
+      "params" : [
+        {
+          "name" : "value",
+          "type" : "long",
+          "optional" : false,
+          "description" : "Values to return"
+        },
+        {
+          "name" : "sort",
+          "type" : "date",
+          "optional" : false,
+          "description" : "Sort key"
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "long"
+    },
+    {
+      "params" : [
+        {
+          "name" : "value",
+          "type" : "long",
+          "optional" : false,
+          "description" : "Values to return"
+        },
+        {
+          "name" : "sort",
+          "type" : "date_nanos",
+          "optional" : false,
+          "description" : "Sort key"
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "long"
+    }
+  ],
+  "examples" : [
+    "FROM k8s\n| STATS first_bytes_in = FIRST(network.bytes_in, @timestamp) BY pod\n| SORT pod ASC"
+  ],
+  "preview" : false,
+  "snapshot_only" : true
+}

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

@@ -0,0 +1,121 @@
+{
+  "comment" : "This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.",
+  "type" : "agg",
+  "name" : "last",
+  "description" : "The latest value of a field.",
+  "signatures" : [
+    {
+      "params" : [
+        {
+          "name" : "value",
+          "type" : "double",
+          "optional" : false,
+          "description" : "Values to return"
+        },
+        {
+          "name" : "sort",
+          "type" : "date",
+          "optional" : false,
+          "description" : "Sort key"
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "double"
+    },
+    {
+      "params" : [
+        {
+          "name" : "value",
+          "type" : "double",
+          "optional" : false,
+          "description" : "Values to return"
+        },
+        {
+          "name" : "sort",
+          "type" : "date_nanos",
+          "optional" : false,
+          "description" : "Sort key"
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "double"
+    },
+    {
+      "params" : [
+        {
+          "name" : "value",
+          "type" : "integer",
+          "optional" : false,
+          "description" : "Values to return"
+        },
+        {
+          "name" : "sort",
+          "type" : "date",
+          "optional" : false,
+          "description" : "Sort key"
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "integer"
+    },
+    {
+      "params" : [
+        {
+          "name" : "value",
+          "type" : "integer",
+          "optional" : false,
+          "description" : "Values to return"
+        },
+        {
+          "name" : "sort",
+          "type" : "date_nanos",
+          "optional" : false,
+          "description" : "Sort key"
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "integer"
+    },
+    {
+      "params" : [
+        {
+          "name" : "value",
+          "type" : "long",
+          "optional" : false,
+          "description" : "Values to return"
+        },
+        {
+          "name" : "sort",
+          "type" : "date",
+          "optional" : false,
+          "description" : "Sort key"
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "long"
+    },
+    {
+      "params" : [
+        {
+          "name" : "value",
+          "type" : "long",
+          "optional" : false,
+          "description" : "Values to return"
+        },
+        {
+          "name" : "sort",
+          "type" : "date_nanos",
+          "optional" : false,
+          "description" : "Sort key"
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "long"
+    }
+  ],
+  "examples" : [
+    "FROM k8s\n| STATS last_bytes_in = LAST(network.bytes_in, @timestamp) BY pod\n| SORT pod ASC"
+  ],
+  "preview" : false,
+  "snapshot_only" : true
+}

docs/reference/query-languages/esql/kibana/docs/functions/first.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
+The earliest value of a field.
+
+```esql
+FROM k8s
+| STATS first_bytes_in = FIRST(network.bytes_in, @timestamp) BY pod
+| SORT pod ASC
+```

docs/reference/query-languages/esql/kibana/docs/functions/last.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
+The latest value of a field.
+
+```esql
+FROM k8s
+| STATS last_bytes_in = LAST(network.bytes_in, @timestamp) BY pod
+| SORT pod ASC
+```

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

@@ -0,0 +1,91 @@
+long_by_date
+required_capability: agg_first_last
+// tag::first[]
+FROM k8s
+| STATS first_bytes_in = FIRST(network.bytes_in, @timestamp) BY pod
+| SORT pod ASC
+// end::first[]
+;
+
+// tag::first-result[]
+first_bytes_in:long | pod:keyword
+                278 | one
+                473 | three
+                699 | two
+// end::first-result[]
+;
+
+double_by_date
+required_capability: agg_first_last
+FROM k8s
+| STATS first_network_cost = FIRST(network.cost, @timestamp) BY pod
+| SORT pod ASC
+;
+
+first_network_cost:double | pod:keyword
+                    5.375 | one
+                    1.250 | three
+                    9.375 | two
+;
+
+by_date_nanos
+required_capability: agg_first_last
+FROM date_nanos
+| STATS first_num = FIRST(num, nanos)
+;
+
+  first_num:long
+               0
+;
+
+
+row
+required_capability: agg_first_last
+ROW a = 100, @timestamp = "2025-01-01"::DATETIME
+| STATS FIRST(a, @timestamp)
+;
+
+FIRST(a, @timestamp):integer
+100
+;
+
+double_by_null
+required_capability: agg_first_last
+FROM k8s
+| EVAL @timestamp = @timestamp + null
+| STATS first_network_cost = FIRST(network.cost, @timestamp) BY pod
+| SORT pod ASC
+;
+
+first_network_cost:double | pod:keyword
+                     null | one
+                     null | three
+                     null | two
+;
+
+null_by_timestamp
+required_capability: agg_first_last
+FROM k8s
+| EVAL network.cost = network.cost + null
+| STATS first_network_cost = FIRST(network.cost, @timestamp) BY pod
+| SORT pod ASC
+;
+
+first_network_cost:double | pod:keyword
+                     null | one
+                     null | three
+                     null | two
+;
+
+skips_null
+required_capability: agg_first_last
+ROW v = ["null", "3", "2"]
+| MV_EXPAND v
+| EVAL v = CASE(v == "null", null, v::INTEGER)
+| EVAL @timestamp = CONCAT("2025-08-0", COALESCE(v::KEYWORD, "1"))::DATE
+| STATS FIRST(v, @timestamp)
+;
+
+FIRST(v, @timestamp):integer
+            2
+;

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

@@ -0,0 +1,91 @@
+long_by_date
+required_capability: agg_first_last
+// tag::last[]
+FROM k8s
+| STATS last_bytes_in = LAST(network.bytes_in, @timestamp) BY pod
+| SORT pod ASC
+// end::last[]
+;
+
+// tag::last-result[]
+last_bytes_in:long | pod:keyword
+               206 | one
+               972 | three
+               812 | two
+// end::last-result[]
+;
+
+double_by_date
+required_capability: agg_first_last
+FROM k8s
+| STATS last_network_cost = LAST(network.cost, @timestamp) BY pod
+| SORT pod ASC
+;
+
+last_network_cost:double | pod:keyword
+                   6.250 | one
+                  10.875 | three
+                  10.750 | two
+;
+
+by_date_nanos
+required_capability: agg_first_last
+FROM date_nanos
+| STATS last_num = LAST(num, nanos)
+;
+
+   last_num:long
+1698069301543123456
+;
+
+
+row
+required_capability: agg_first_last
+ROW a = 100, @timestamp = "2025-01-01"::DATETIME
+| STATS LAST(a, @timestamp)
+;
+
+LAST(a, @timestamp):integer
+100
+;
+
+double_by_null
+required_capability: agg_first_last
+FROM k8s
+| EVAL @timestamp = @timestamp + null
+| STATS last_network_cost = LAST(network.cost, @timestamp) BY pod
+| SORT pod ASC
+;
+
+last_network_cost:double | pod:keyword
+                    null | one
+                    null | three
+                    null | two
+;
+
+null_by_timestamp
+required_capability: agg_first_last
+FROM k8s
+| EVAL network.cost = network.cost + null
+| STATS last_network_cost = LAST(network.cost, @timestamp) BY pod
+| SORT pod ASC
+;
+
+last_network_cost:double | pod:keyword
+                    null | one
+                    null | three
+                    null | two
+;
+
+skips_null
+required_capability: agg_first_last
+ROW v = ["null", "1", "2"]
+| MV_EXPAND v
+| EVAL v = CASE(v == "null", null, v::INTEGER)
+| EVAL @timestamp = CONCAT("2025-08-0", COALESCE(v::KEYWORD, "3"))::DATE
+| STATS LAST(v, @timestamp)
+;
+
+LAST(v, @timestamp):integer
+            2
+;

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

@@ -1337,6 +1337,11 @@ public class EsqlCapabilities {
          */
         CATEGORIZE_OPTIONS,
 
+        /**
+         * FIRST and LAST aggregate functions.
+         */
+        AGG_FIRST_LAST(Build.current().isSnapshot()),
+
         /**
          * Support correct counting of skipped shards.
          */

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

@@ -25,7 +25,9 @@ import org.elasticsearch.xpack.esql.expression.function.aggregate.Count;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.CountDistinct;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.CountDistinctOverTime;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.CountOverTime;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.First;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.FirstOverTime;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.Last;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.LastOverTime;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Max;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.MaxOverTime;
@@ -475,6 +477,8 @@ 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(First.class, bi(First::new), "first"),
+                def(Last.class, bi(Last::new), "last"),
                 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"),

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

@@ -18,6 +18,7 @@ public class AggregateWritables {
             Avg.ENTRY,
             Count.ENTRY,
             CountDistinct.ENTRY,
+            First.ENTRY,
             Max.ENTRY,
             Median.ENTRY,
             MedianAbsoluteDeviation.ENTRY,
@@ -34,6 +35,7 @@ public class AggregateWritables {
             MinOverTime.ENTRY,
             MaxOverTime.ENTRY,
             AvgOverTime.ENTRY,
+            Last.ENTRY,
             LastOverTime.ENTRY,
             FirstOverTime.ENTRY,
             SumOverTime.ENTRY,

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

@@ -0,0 +1,129 @@
+/*
+ * 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.compute.aggregation.AggregatorFunctionSupplier;
+import org.elasticsearch.compute.aggregation.FirstDoubleByTimestampAggregatorFunctionSupplier;
+import org.elasticsearch.compute.aggregation.FirstFloatByTimestampAggregatorFunctionSupplier;
+import org.elasticsearch.compute.aggregation.FirstIntByTimestampAggregatorFunctionSupplier;
+import org.elasticsearch.compute.aggregation.FirstLongByTimestampAggregatorFunctionSupplier;
+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.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.Param;
+import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
+import org.elasticsearch.xpack.esql.planner.ToAggregator;
+
+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 First extends AggregateFunction implements ToAggregator {
+    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(Expression.class, "First", First::readFrom);
+
+    private final Expression sort;
+
+    // TODO: support all types of values
+    @FunctionInfo(
+        type = FunctionType.AGGREGATE,
+        returnType = { "long", "integer", "double" },
+        description = "The earliest value of a field.",
+        appliesTo = @FunctionAppliesTo(lifeCycle = FunctionAppliesToLifecycle.UNAVAILABLE),
+        examples = @Example(file = "stats_first", tag = "first")
+    )
+    public First(
+        Source source,
+        @Param(name = "value", type = { "long", "integer", "double" }, description = "Values to return") Expression field,
+        @Param(name = "sort", type = { "date", "date_nanos" }, description = "Sort key") Expression sort
+    ) {
+        this(source, field, Literal.TRUE, sort);
+    }
+
+    private First(Source source, Expression field, Expression filter, Expression sort) {
+        super(source, field, filter, List.of(sort));
+        this.sort = sort;
+    }
+
+    private static First readFrom(StreamInput in) throws IOException {
+        Source source = Source.readFrom((PlanStreamInput) in);
+        Expression field = in.readNamedWriteable(Expression.class);
+        Expression filter = in.readNamedWriteable(Expression.class);
+        List<Expression> params = in.readNamedWriteableCollectionAsList(Expression.class);
+        Expression sort = params.getFirst();
+        return new First(source, field, filter, sort);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return ENTRY.name;
+    }
+
+    @Override
+    protected NodeInfo<First> info() {
+        return NodeInfo.create(this, First::new, field(), sort);
+    }
+
+    @Override
+    public First replaceChildren(List<Expression> newChildren) {
+        return new First(source(), newChildren.get(0), newChildren.get(1), newChildren.get(2));
+    }
+
+    @Override
+    public First withFilter(Expression filter) {
+        return new First(source(), field(), filter, sort);
+    }
+
+    @Override
+    public DataType dataType() {
+        return field().dataType();
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        return isType(field(), dt -> dt.isNumeric() && dt != DataType.UNSIGNED_LONG, sourceText(), DEFAULT, "numeric except unsigned_long")
+            .and(
+                isType(
+                    sort,
+                    dt -> dt == DataType.LONG || dt == DataType.DATETIME || dt == DataType.DATE_NANOS,
+                    sourceText(),
+                    SECOND,
+                    "long or date_nanos or datetime"
+                )
+            );
+    }
+
+    @Override
+    public AggregatorFunctionSupplier supplier() {
+        final DataType type = field().dataType();
+        return switch (type) {
+            case LONG -> new FirstLongByTimestampAggregatorFunctionSupplier();
+            case INTEGER -> new FirstIntByTimestampAggregatorFunctionSupplier();
+            case DOUBLE -> new FirstDoubleByTimestampAggregatorFunctionSupplier();
+            case FLOAT -> new FirstFloatByTimestampAggregatorFunctionSupplier();
+            default -> throw EsqlIllegalArgumentException.illegalDataType(type);
+        };
+    }
+
+    @Override
+    public String toString() {
+        return "first(" + field() + ", " + sort + ")";
+    }
+}

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

@@ -0,0 +1,129 @@
+/*
+ * 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.compute.aggregation.AggregatorFunctionSupplier;
+import org.elasticsearch.compute.aggregation.LastDoubleByTimestampAggregatorFunctionSupplier;
+import org.elasticsearch.compute.aggregation.LastFloatByTimestampAggregatorFunctionSupplier;
+import org.elasticsearch.compute.aggregation.LastIntByTimestampAggregatorFunctionSupplier;
+import org.elasticsearch.compute.aggregation.LastLongByTimestampAggregatorFunctionSupplier;
+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.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.Param;
+import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
+import org.elasticsearch.xpack.esql.planner.ToAggregator;
+
+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 Last extends AggregateFunction implements ToAggregator {
+    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(Expression.class, "Last", Last::readFrom);
+
+    private final Expression sort;
+
+    // TODO: support all types
+    @FunctionInfo(
+        type = FunctionType.AGGREGATE,
+        returnType = { "long", "integer", "double" },
+        description = "The latest value of a field.",
+        appliesTo = @FunctionAppliesTo(lifeCycle = FunctionAppliesToLifecycle.UNAVAILABLE),
+        examples = @Example(file = "stats_last", tag = "last")
+    )
+    public Last(
+        Source source,
+        @Param(name = "value", type = { "long", "integer", "double" }, description = "Values to return") Expression field,
+        @Param(name = "sort", type = { "date", "date_nanos" }, description = "Sort key") Expression sort
+    ) {
+        this(source, field, Literal.TRUE, sort);
+    }
+
+    private Last(Source source, Expression field, Expression filter, Expression sort) {
+        super(source, field, filter, List.of(sort));
+        this.sort = sort;
+    }
+
+    private static Last readFrom(StreamInput in) throws IOException {
+        Source source = Source.readFrom((PlanStreamInput) in);
+        Expression field = in.readNamedWriteable(Expression.class);
+        Expression filter = in.readNamedWriteable(Expression.class);
+        List<Expression> params = in.readNamedWriteableCollectionAsList(Expression.class);
+        Expression sort = params.getFirst();
+        return new Last(source, field, filter, sort);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return ENTRY.name;
+    }
+
+    @Override
+    protected NodeInfo<Last> info() {
+        return NodeInfo.create(this, Last::new, field(), sort);
+    }
+
+    @Override
+    public Last replaceChildren(List<Expression> newChildren) {
+        return new Last(source(), newChildren.get(0), newChildren.get(1), newChildren.get(2));
+    }
+
+    @Override
+    public Last withFilter(Expression filter) {
+        return new Last(source(), field(), filter, sort);
+    }
+
+    @Override
+    public DataType dataType() {
+        return field().dataType();
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        return isType(field(), dt -> dt.isNumeric() && dt != DataType.UNSIGNED_LONG, sourceText(), DEFAULT, "numeric except unsigned_long")
+            .and(
+                isType(
+                    sort,
+                    dt -> dt == DataType.LONG || dt == DataType.DATETIME || dt == DataType.DATE_NANOS,
+                    sourceText(),
+                    SECOND,
+                    "long or date_nanos or datetime"
+                )
+            );
+    }
+
+    @Override
+    public AggregatorFunctionSupplier supplier() {
+        final DataType type = field().dataType();
+        return switch (type) {
+            case LONG -> new LastLongByTimestampAggregatorFunctionSupplier();
+            case INTEGER -> new LastIntByTimestampAggregatorFunctionSupplier();
+            case DOUBLE -> new LastDoubleByTimestampAggregatorFunctionSupplier();
+            case FLOAT -> new LastFloatByTimestampAggregatorFunctionSupplier();
+            default -> throw EsqlIllegalArgumentException.illegalDataType(type);
+        };
+    }
+
+    @Override
+    public String toString() {
+        return "last(" + field() + ", " + sort + ")";
+    }
+}

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

@@ -26,6 +26,7 @@ import java.util.function.Supplier;
 
 import static org.elasticsearch.test.ESTestCase.randomBoolean;
 import static org.elasticsearch.test.ESTestCase.randomList;
+import static org.elasticsearch.xpack.esql.core.util.NumericUtils.UNSIGNED_LONG_MAX;
 import static org.elasticsearch.xpack.esql.core.util.SpatialCoordinateTypes.GEO;
 import static org.elasticsearch.xpack.esql.expression.function.TestCaseSupplier.TypedDataSupplier;
 
@@ -36,6 +37,22 @@ public final class MultiRowTestCaseSupplier {
 
     private MultiRowTestCaseSupplier() {}
 
+    /**
+     * A {@link List} of the cases for the specified type without any limits.
+     */
+    public static List<TypedDataSupplier> unlimitedSuppliers(DataType type, int minRows, int maxRows) {
+        return switch (type) {
+            case DATETIME -> dateCases(minRows, maxRows);
+            case DATE_NANOS -> dateNanosCases(minRows, maxRows);
+            case INTEGER -> intCases(minRows, maxRows, Integer.MIN_VALUE, Integer.MAX_VALUE, true);
+            case LONG -> longCases(minRows, maxRows, Long.MIN_VALUE, Long.MAX_VALUE, true);
+            case UNSIGNED_LONG -> ulongCases(minRows, maxRows, BigInteger.ZERO, UNSIGNED_LONG_MAX, true);
+            case DOUBLE -> doubleCases(minRows, maxRows, -Double.MAX_VALUE, Double.MAX_VALUE, true);
+            // If a type is missing here it's safe to them as you need them
+            default -> throw new IllegalArgumentException("unsupported type [" + type + "]");
+        };
+    }
+
     public static List<TypedDataSupplier> nullCases(int minRows, int maxRows) {
         List<TypedDataSupplier> cases = new ArrayList<>();
 

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

@@ -0,0 +1,91 @@
+/*
+ * 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.common.collect.Iterators;
+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.TestCaseSupplier;
+import org.hamcrest.Matchers;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Supplier;
+
+import static org.elasticsearch.xpack.esql.expression.function.MultiRowTestCaseSupplier.unlimitedSuppliers;
+import static org.hamcrest.Matchers.anyOf;
+
+public class FirstTests extends AbstractAggregationTestCase {
+    public FirstTests(@Name("TestCase") Supplier<TestCaseSupplier.TestCase> testCaseSupplier) {
+        this.testCase = testCaseSupplier.get();
+    }
+
+    @ParametersFactory
+    public static Iterable<Object[]> parameters() {
+        int rows = 1000;
+        List<TestCaseSupplier> suppliers = new ArrayList<>();
+
+        for (DataType valueType : List.of(DataType.INTEGER, DataType.LONG, DataType.DOUBLE)) {
+            for (TestCaseSupplier.TypedDataSupplier valueSupplier : unlimitedSuppliers(valueType, rows, rows)) {
+                for (DataType sortType : List.of(DataType.DATETIME, DataType.DATE_NANOS)) {
+                    for (TestCaseSupplier.TypedDataSupplier sortSupplier : unlimitedSuppliers(sortType, rows, rows)) {
+                        suppliers.add(makeSupplier(valueSupplier, sortSupplier, true));
+                    }
+                }
+            }
+        }
+        return parameterSuppliersFromTypedData(randomizeBytesRefsOffset(suppliers));
+    }
+
+    @Override
+    protected Expression build(Source source, List<Expression> args) {
+        return new First(source, args.get(0), args.get(1));
+    }
+
+    static TestCaseSupplier makeSupplier(
+        TestCaseSupplier.TypedDataSupplier valueSupplier,
+        TestCaseSupplier.TypedDataSupplier sortSupplier,
+        boolean first
+    ) {
+        return new TestCaseSupplier(
+            valueSupplier.name() + ", " + sortSupplier.name(),
+            List.of(valueSupplier.type(), sortSupplier.type()),
+            () -> {
+                Long firstSort = null;
+                Set<Object> expected = new HashSet<>();
+                TestCaseSupplier.TypedData values = valueSupplier.get();
+                TestCaseSupplier.TypedData sorts = sortSupplier.get();
+                List<?> valuesList = (List<?>) values.data();
+                List<?> sortsList = (List<?>) sorts.data();
+                for (int p = 0; p < valuesList.size(); p++) {
+                    Long s = (Long) sortsList.get(p);
+                    if (firstSort == null || (first ? s < firstSort : s > firstSort)) {
+                        firstSort = s;
+                        expected.clear();
+                        expected.add(valuesList.get(p));
+                    } else if (firstSort.equals(s)) {
+                        expected.add(valuesList.get(p));
+                    }
+                }
+                return new TestCaseSupplier.TestCase(
+                    List.of(values, sorts),
+                    "unused",
+                    values.type(),
+                    anyOf(() -> Iterators.map(expected.iterator(), Matchers::equalTo))
+                );
+            }
+        );
+    }
+}

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

@@ -0,0 +1,52 @@
+/*
+ * 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.TestCaseSupplier;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Supplier;
+
+import static org.elasticsearch.xpack.esql.expression.function.MultiRowTestCaseSupplier.unlimitedSuppliers;
+import static org.elasticsearch.xpack.esql.expression.function.aggregate.FirstTests.makeSupplier;
+
+public class LastTests extends AbstractAggregationTestCase {
+    public LastTests(@Name("TestCase") Supplier<TestCaseSupplier.TestCase> testCaseSupplier) {
+        this.testCase = testCaseSupplier.get();
+    }
+
+    @ParametersFactory
+    public static Iterable<Object[]> parameters() {
+        int rows = 1000;
+        List<TestCaseSupplier> suppliers = new ArrayList<>();
+
+        for (DataType valueType : List.of(DataType.INTEGER, DataType.LONG, DataType.DOUBLE)) {
+            for (TestCaseSupplier.TypedDataSupplier valueSupplier : unlimitedSuppliers(valueType, rows, rows)) {
+                for (DataType sortType : List.of(DataType.DATETIME, DataType.DATE_NANOS)) {
+                    for (TestCaseSupplier.TypedDataSupplier sortSupplier : unlimitedSuppliers(sortType, rows, rows)) {
+                        suppliers.add(makeSupplier(valueSupplier, sortSupplier, false));
+                    }
+                }
+            }
+        }
+        return parameterSuppliersFromTypedData(randomizeBytesRefsOffset(suppliers));
+    }
+
+    @Override
+    protected Expression build(Source source, List<Expression> args) {
+        return new Last(source, args.get(0), args.get(1));
+    }
+}