ES|QL PresentOverTime function (#134355)

* ES|QL: Add PRESENT ES|QL function

Add a new ES|QL function that checks for the presence of a field over time. 
Presence means that the input expression yields any non-null value.

Part of #131069

docs/reference/query-languages/esql/_snippets/functions/description/present_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 presence of a field in the output result over time range.
+
+::::{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/present.md

@@ -27,4 +27,16 @@ FROM employees
 | true | 5 |
 | true | null |
 
+To check for the presence and return 1 when it's true and 0 when it's false
+
+```esql
+FROM employees
+| WHERE emp_no == 10020
+| STATS is_present = TO_INTEGER(PRESENT(languages))
+```
+
+| is_present:integer |
+| --- |
+| 0 |
+
 

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

@@ -0,0 +1,18 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Example**
+
+```esql
+TS k8s
+| WHERE cluster == "prod" AND pod == "two"
+| STATS events_received = max(present_over_time(events_received)) BY pod, time_bucket = tbucket(2 minute)
+```
+
+| events_received:boolean | pod:keyword | time_bucket:datetime |
+| --- | --- | --- |
+| true | two | 2024-05-10T00:02:00.000Z |
+| true | two | 2024-05-10T00:08:00.000Z |
+| false | two | 2024-05-10T00:10:00.000Z |
+| false | two | 2024-05-10T00:12:00.000Z |
+
+

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

docs/reference/query-languages/esql/_snippets/functions/parameters/present_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/types/present_over_time.md

@@ -0,0 +1,25 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+**Supported types**
+
+| field | result |
+| --- | --- |
+| boolean | boolean |
+| cartesian_point | boolean |
+| cartesian_shape | boolean |
+| date | boolean |
+| date_nanos | boolean |
+| double | boolean |
+| geo_point | boolean |
+| geo_shape | boolean |
+| geohash | boolean |
+| geohex | boolean |
+| geotile | boolean |
+| integer | boolean |
+| ip | boolean |
+| keyword | boolean |
+| long | boolean |
+| text | boolean |
+| unsigned_long | boolean |
+| version | boolean |
+

docs/reference/query-languages/esql/images/functions/present_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="408" height="46" viewbox="0 0 408 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 31h5m224 0h10m32 0h10m80 0h10m32 0h5"/><rect class="s" x="5" y="5" width="224" height="36"/><text class="k" x="15" y="31">PRESENT_OVER_TIME</text><rect class="s" x="239" y="5" width="32" height="36" rx="7"/><text class="syn" x="249" y="31">(</text><rect class="s" x="281" y="5" width="80" height="36" rx="7"/><text class="k" x="291" y="31">field</text><rect class="s" x="371" y="5" width="32" height="36" rx="7"/><text class="syn" x="381" y="31">)</text></svg>

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

@@ -223,7 +223,8 @@
   ],
   "examples" : [
     "FROM employees\n| STATS is_present = PRESENT(languages)",
-    "FROM employees\n| STATS is_present = PRESENT(salary) BY languages"
+    "FROM employees\n| STATS is_present = PRESENT(salary) BY languages",
+    "FROM employees\n| WHERE emp_no == 10020\n| STATS is_present = TO_INTEGER(PRESENT(languages))"
   ],
   "preview" : false,
   "snapshot_only" : false

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

@@ -0,0 +1,230 @@
+{
+  "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" : "present_over_time",
+  "description" : "The presence of a field in the output result over time range.",
+  "note" : "Available with the TS command in snapshot builds",
+  "signatures" : [
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "boolean",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "cartesian_point",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "cartesian_shape",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "date",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "date_nanos",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "double",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "geo_point",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "geo_shape",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "geohash",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "geohex",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "geotile",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "integer",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "ip",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "keyword",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "long",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "text",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "unsigned_long",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    },
+    {
+      "params" : [
+        {
+          "name" : "field",
+          "type" : "version",
+          "optional" : false,
+          "description" : ""
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "boolean"
+    }
+  ],
+  "examples" : [
+    "TS k8s\n| WHERE cluster == \"prod\" AND pod == \"two\"\n| STATS events_received = max(present_over_time(events_received)) BY pod, time_bucket = tbucket(2 minute)"
+  ],
+  "preview" : false,
+  "snapshot_only" : true
+}

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

@@ -0,0 +1,12 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do not edit it. See ../README.md for how to regenerate it.
+
+### PRESENT OVER TIME
+The presence of a field in the output result over time range.
+
+Note: Available with the [TS](https://www.elastic.co/docs/reference/query-languages/esql/commands/source-commands#esql-ts) command in snapshot builds
+
+```esql
+TS k8s
+| WHERE cluster == "prod" AND pod == "two"
+| STATS events_received = max(present_over_time(events_received)) BY pod, time_bucket = tbucket(2 minute)
+```

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

@@ -0,0 +1,323 @@
+present_over_time_events_received
+required_capability: metrics_command
+required_capability: present_over_time
+
+TS k8s 
+| WHERE cluster == "prod" AND pod == "two" 
+| STATS events_received = max(present_over_time(events_received)) BY pod, time_bucket = tbucket(2 minute) 
+| SORT time_bucket
+;
+ignoreOrder:true
+
+events_received:boolean | pod:keyword | time_bucket:datetime
+true                    | two         | 2024-05-10T00:02:00.000Z
+true                    | two         | 2024-05-10T00:08:00.000Z
+false                   | two         | 2024-05-10T00:10:00.000Z
+false                   | two         | 2024-05-10T00:12:00.000Z
+true                    | two         | 2024-05-10T00:14:00.000Z
+true                    | two         | 2024-05-10T00:16:00.000Z
+false                   | two         | 2024-05-10T00:18:00.000Z
+true                    | two         | 2024-05-10T00:20:00.000Z
+true                    | two         | 2024-05-10T00:22:00.000Z
+;
+
+present_over_time_of_long
+required_capability: metrics_command
+required_capability: present_over_time
+TS k8s | STATS is_present = max(present_over_time(network.bytes_in)) BY cluster, time_bucket = tbucket(10minute) | SORT cluster, time_bucket | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true               | prod            | 2024-05-10T00:00:00.000Z
+true               | prod            | 2024-05-10T00:10:00.000Z
+true               | prod            | 2024-05-10T00:20:00.000Z
+true               | qa              | 2024-05-10T00:00:00.000Z
+true               | qa              | 2024-05-10T00:10:00.000Z
+true               | qa              | 2024-05-10T00:20:00.000Z
+true               | staging         | 2024-05-10T00:00:00.000Z
+true               | staging         | 2024-05-10T00:10:00.000Z
+true               | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_of_boolean
+required_capability: metrics_command
+required_capability: present_over_time
+required_capability: k8s_dataset_additional_fields
+TS k8s | STATS is_present = max(present_over_time(network.eth0.up)) BY cluster, time_bucket = tbucket(10minute) | SORT time_bucket, cluster | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true          | prod            | 2024-05-10T00:00:00.000Z
+true          | qa              | 2024-05-10T00:00:00.000Z
+true          | staging         | 2024-05-10T00:00:00.000Z
+true          | prod            | 2024-05-10T00:10:00.000Z
+true          | qa              | 2024-05-10T00:10:00.000Z
+true          | staging         | 2024-05-10T00:10:00.000Z
+true          | prod            | 2024-05-10T00:20:00.000Z
+true          | qa              | 2024-05-10T00:20:00.000Z
+true          | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_of_date_nanos
+required_capability: metrics_command
+required_capability: present_over_time
+required_capability: k8s_dataset_additional_fields
+TS k8s | STATS is_present = max(present_over_time(network.eth0.last_up)) BY cluster, time_bucket = tbucket(10minute) | SORT time_bucket, cluster | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true           | prod            | 2024-05-10T00:00:00.000Z
+true           | qa              | 2024-05-10T00:00:00.000Z
+true           | staging         | 2024-05-10T00:00:00.000Z
+true           | prod            | 2024-05-10T00:10:00.000Z
+true           | qa              | 2024-05-10T00:10:00.000Z
+true           | staging         | 2024-05-10T00:10:00.000Z
+true           | prod            | 2024-05-10T00:20:00.000Z
+true           | qa              | 2024-05-10T00:20:00.000Z
+true           | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_of_date
+required_capability: metrics_command
+required_capability: present_over_time
+required_capability: k8s_dataset_additional_fields
+TS k8s | STATS is_present = max(present_over_time(to_datetime(network.eth0.last_up))) BY cluster, time_bucket = tbucket(10minute) | SORT time_bucket, cluster | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true           | prod            | 2024-05-10T00:00:00.000Z
+true           | qa              | 2024-05-10T00:00:00.000Z
+true           | staging         | 2024-05-10T00:00:00.000Z
+true           | prod            | 2024-05-10T00:10:00.000Z
+true           | qa              | 2024-05-10T00:10:00.000Z
+true           | staging         | 2024-05-10T00:10:00.000Z
+true           | prod            | 2024-05-10T00:20:00.000Z
+true           | qa              | 2024-05-10T00:20:00.000Z
+true           | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_of_version
+required_capability: metrics_command
+required_capability: present_over_time
+required_capability: k8s_dataset_additional_fields
+TS k8s | STATS is_present = max(present_over_time(network.eth0.firmware_version)) BY cluster, time_bucket = tbucket(10minute) | SORT time_bucket, cluster | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true           | prod            | 2024-05-10T00:00:00.000Z
+true           | qa              | 2024-05-10T00:00:00.000Z
+true           | staging         | 2024-05-10T00:00:00.000Z
+true           | prod            | 2024-05-10T00:10:00.000Z
+true           | qa              | 2024-05-10T00:10:00.000Z
+true           | staging         | 2024-05-10T00:10:00.000Z
+true           | prod            | 2024-05-10T00:20:00.000Z
+true           | qa              | 2024-05-10T00:20:00.000Z
+true           | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_of_integer
+required_capability: metrics_command
+required_capability: present_over_time
+required_capability: k8s_dataset_additional_fields
+TS k8s | STATS is_present = max(present_over_time(network.eth0.currently_connected_clients)) BY cluster, time_bucket = tbucket(10minute) | SORT time_bucket, cluster | LIMIT 10;
+
+is_present:boolean     | cluster:keyword | time_bucket:datetime
+true  | prod            | 2024-05-10T00:00:00.000Z
+true | qa              | 2024-05-10T00:00:00.000Z
+true                | staging         | 2024-05-10T00:00:00.000Z
+true | prod            | 2024-05-10T00:10:00.000Z
+true               | qa              | 2024-05-10T00:10:00.000Z
+true  | staging         | 2024-05-10T00:10:00.000Z
+true | prod            | 2024-05-10T00:20:00.000Z
+true                | qa              | 2024-05-10T00:20:00.000Z
+true | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_of_text
+required_capability: metrics_command
+required_capability: present_over_time
+TS k8s | STATS is_present = max(present_over_time(event_log)) BY cluster, time_bucket = tbucket(10minute) | SORT cluster, time_bucket | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true               | prod            | 2024-05-10T00:00:00.000Z
+true               | prod            | 2024-05-10T00:10:00.000Z
+true               | prod            | 2024-05-10T00:20:00.000Z
+true               | qa              | 2024-05-10T00:00:00.000Z
+true               | qa              | 2024-05-10T00:10:00.000Z
+true               | qa              | 2024-05-10T00:20:00.000Z
+true               | staging         | 2024-05-10T00:00:00.000Z
+true               | staging         | 2024-05-10T00:10:00.000Z
+true               | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_of_keyword
+required_capability: metrics_command
+required_capability: present_over_time
+required_capability: k8s_dataset_additional_fields
+TS k8s | STATS is_present = max(present_over_time(pod)) BY cluster, time_bucket = tbucket(10minute) | SORT time_bucket, cluster | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true     | prod            | 2024-05-10T00:00:00.000Z
+true     | qa              | 2024-05-10T00:00:00.000Z
+true     | staging         | 2024-05-10T00:00:00.000Z
+true     | prod            | 2024-05-10T00:10:00.000Z
+true     | qa              | 2024-05-10T00:10:00.000Z
+true     | staging         | 2024-05-10T00:10:00.000Z
+true     | prod            | 2024-05-10T00:20:00.000Z
+true     | qa              | 2024-05-10T00:20:00.000Z
+true     | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_of_aggregate_metric_double
+required_capability: metrics_command
+required_capability: present_over_time
+TS k8s-downsampled | STATS is_present = max(present_over_time(network.eth0.tx)) BY cluster, time_bucket = tbucket(10 minute) | SORT time_bucket, cluster | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true      | prod            | 2024-05-09T23:30:00.000Z
+true      | qa              | 2024-05-09T23:30:00.000Z
+true      | staging         | 2024-05-09T23:30:00.000Z
+true      | prod            | 2024-05-09T23:40:00.000Z
+true      | qa              | 2024-05-09T23:40:00.000Z
+true      | staging         | 2024-05-09T23:40:00.000Z
+true      | prod            | 2024-05-09T23:50:00.000Z
+true      | qa              | 2024-05-09T23:50:00.000Z
+true      | staging         | 2024-05-09T23:50:00.000Z
+;
+
+present_over_time_of_geopoint
+required_capability: metrics_command
+required_capability: present_over_time
+required_capability: k8s_datasets_geospatial_fields
+TS k8s | STATS is_present = max(present_over_time(event_city)) BY cluster, time_bucket = tbucket(10minute) | SORT time_bucket, cluster | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true       | prod            | 2024-05-10T00:00:00.000Z
+true       | qa              | 2024-05-10T00:00:00.000Z
+true       | staging         | 2024-05-10T00:00:00.000Z
+true       | prod            | 2024-05-10T00:10:00.000Z
+true       | qa              | 2024-05-10T00:10:00.000Z
+true       | staging         | 2024-05-10T00:10:00.000Z
+true       | prod            | 2024-05-10T00:20:00.000Z
+true       | qa              | 2024-05-10T00:20:00.000Z
+true       | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_of_geoshape
+required_capability: metrics_command
+required_capability: present_over_time
+required_capability: k8s_datasets_geospatial_fields
+TS k8s | STATS is_present = max(present_over_time(event_city_boundary)) BY cluster, time_bucket = tbucket(10minute) | SORT time_bucket, cluster | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true       | prod            | 2024-05-10T00:00:00.000Z
+true       | qa              | 2024-05-10T00:00:00.000Z
+true       | staging         | 2024-05-10T00:00:00.000Z
+true       | prod            | 2024-05-10T00:10:00.000Z
+true       | qa              | 2024-05-10T00:10:00.000Z
+true       | staging         | 2024-05-10T00:10:00.000Z
+true       | prod            | 2024-05-10T00:20:00.000Z
+true       | qa              | 2024-05-10T00:20:00.000Z
+true       | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_of_shape
+required_capability: metrics_command
+required_capability: present_over_time
+required_capability: k8s_datasets_geospatial_fields
+TS k8s | STATS is_present = max(present_over_time(event_shape)) BY cluster, time_bucket = tbucket(10minute) | SORT time_bucket, cluster | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true       | prod            | 2024-05-10T00:00:00.000Z
+true       | qa              | 2024-05-10T00:00:00.000Z
+true       | staging         | 2024-05-10T00:00:00.000Z
+true       | prod            | 2024-05-10T00:10:00.000Z
+true       | qa              | 2024-05-10T00:10:00.000Z
+true       | staging         | 2024-05-10T00:10:00.000Z
+true       | prod            | 2024-05-10T00:20:00.000Z
+true       | qa              | 2024-05-10T00:20:00.000Z
+true       | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_of_point
+required_capability: metrics_command
+required_capability: present_over_time
+required_capability: k8s_datasets_geospatial_fields
+TS k8s | STATS is_present = max(present_over_time(event_location)) BY cluster, time_bucket = tbucket(10minute) | SORT time_bucket, cluster | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true       | prod            | 2024-05-10T00:00:00.000Z
+true       | qa              | 2024-05-10T00:00:00.000Z
+true       | staging         | 2024-05-10T00:00:00.000Z
+true       | prod            | 2024-05-10T00:10:00.000Z
+true       | qa              | 2024-05-10T00:10:00.000Z
+true       | staging         | 2024-05-10T00:10:00.000Z
+true       | prod            | 2024-05-10T00:20:00.000Z
+true       | qa              | 2024-05-10T00:20:00.000Z
+true       | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_with_filtering
+required_capability: metrics_command
+required_capability: present_over_time
+TS k8s | WHERE pod != "three" | STATS is_present = max(present_over_time(network.bytes_in)) BY cluster, time_bucket = tbucket(10 minute) | SORT time_bucket, cluster | LIMIT 10;
+
+is_present:boolean | cluster:keyword | time_bucket:datetime
+true      | prod            | 2024-05-10T00:00:00.000Z
+true      | qa              | 2024-05-10T00:00:00.000Z
+true      | staging         | 2024-05-10T00:00:00.000Z
+true      | prod            | 2024-05-10T00:10:00.000Z
+true      | qa              | 2024-05-10T00:10:00.000Z
+true      | staging         | 2024-05-10T00:10:00.000Z
+true      | prod            | 2024-05-10T00:20:00.000Z
+true      | qa              | 2024-05-10T00:20:00.000Z
+true      | staging         | 2024-05-10T00:20:00.000Z
+;
+
+present_over_time_older_than_10d
+required_capability: metrics_command
+required_capability: present_over_time
+TS k8s-downsampled | WHERE cluster == "qa" AND @timestamp < now() - 10 day | STATS is_present = max(present_over_time(network.eth0.rx)) BY pod, time_bucket = tbucket(10 minute) | SORT time_bucket, pod | LIMIT 5;
+
+is_present:boolean | pod:keyword | time_bucket:datetime
+true       | one         | 2024-05-09T23:30:00.000Z
+true       | three       | 2024-05-09T23:30:00.000Z
+true       | two         | 2024-05-09T23:30:00.000Z
+true       | one         | 2024-05-09T23:40:00.000Z
+true       | three       | 2024-05-09T23:40:00.000Z
+;
+
+eval_on_present_over_time
+required_capability: metrics_command
+required_capability: present_over_time
+TS k8s | STATS is_present = max(present_over_time(network.bytes_in)) BY pod, time_bucket = tbucket(10 minute) | EVAL int = to_integer(is_present) | LIMIT 10 | SORT time_bucket, pod;
+
+is_present:boolean | pod:keyword | time_bucket:datetime     | int:integer
+true             | one         | 2024-05-10T00:00:00.000Z | 1
+true             | three       | 2024-05-10T00:00:00.000Z | 1
+true             | two         | 2024-05-10T00:00:00.000Z | 1
+true             | one         | 2024-05-10T00:10:00.000Z | 1
+true             | three       | 2024-05-10T00:10:00.000Z | 1
+true             | two         | 2024-05-10T00:10:00.000Z | 1
+true             | one         | 2024-05-10T00:20:00.000Z | 1
+true             | three       | 2024-05-10T00:20:00.000Z | 1
+true             | two         | 2024-05-10T00:20:00.000Z | 1
+;
+
+present_over_time_events_received_as_integer
+required_capability: metrics_command
+required_capability: present_over_time
+
+TS k8s 
+| WHERE cluster == "prod" AND pod == "two" 
+| STATS events_received = max(to_integer(present_over_time(events_received))) BY pod, time_bucket = tbucket(2 minute) 
+| SORT time_bucket
+;
+ignoreOrder:true
+
+events_received:integer | pod:keyword | time_bucket:datetime
+1                       | two         | 2024-05-10T00:02:00.000Z
+1                       | two         | 2024-05-10T00:08:00.000Z
+0                       | two         | 2024-05-10T00:10:00.000Z
+0                       | two         | 2024-05-10T00:12:00.000Z
+1                       | two         | 2024-05-10T00:14:00.000Z
+1                       | two         | 2024-05-10T00:16:00.000Z
+0                       | two         | 2024-05-10T00:18:00.000Z
+1                       | two         | 2024-05-10T00:20:00.000Z
+1                       | two         | 2024-05-10T00:22:00.000Z
+;

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

@@ -446,3 +446,30 @@ cost_per_mb:double | cluster:keyword | time_bucket:datetime
 2.071474095190272  | prod            | 2024-05-10T00:15:00.000Z
 1.59556462585034   | staging         | 2024-05-10T00:10:00.000Z
 ;
+
+present_over_time
+required_capability: metrics_command
+required_capability: present_over_time
+
+// tag::present_over_time[]
+TS k8s 
+| WHERE cluster == "prod" AND pod == "two" 
+| STATS events_received = max(present_over_time(events_received)) BY pod, time_bucket = tbucket(2 minute)
+// end::present_over_time[] 
+| SORT time_bucket
+;
+ignoreOrder:true
+
+// tag::present_over_time-result[]
+events_received:boolean | pod:keyword | time_bucket:datetime
+true                    | two         | 2024-05-10T00:02:00.000Z
+true                    | two         | 2024-05-10T00:08:00.000Z
+false                   | two         | 2024-05-10T00:10:00.000Z
+false                   | two         | 2024-05-10T00:12:00.000Z
+// end::present_over_time-result[]
+true                    | two         | 2024-05-10T00:14:00.000Z
+true                    | two         | 2024-05-10T00:16:00.000Z
+false                   | two         | 2024-05-10T00:18:00.000Z
+true                    | two         | 2024-05-10T00:20:00.000Z
+true                    | two         | 2024-05-10T00:22:00.000Z
+;

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

@@ -110,3 +110,43 @@ FROM employees
 p_true:boolean    | p_false:boolean
 true              | false
 ;
+
+PRESENT with per-agg filter with toInteger
+required_capability: fn_present
+
+FROM employees
+| STATS p_true = TO_INTEGER(PRESENT(salary)) WHERE gender == "M",
+        p_false = TO_INTEGER(PRESENT(salary)) WHERE gender == "X"
+;
+
+p_true:integer    | p_false:integer
+1                 | 0
+;
+
+TO_INTEGER(PRESENT) with filter returns 1
+required_capability: fn_present
+
+FROM employees
+| WHERE emp_no IN (10019, 10020)
+| STATS is_present = TO_INTEGER(PRESENT(languages))
+;
+
+is_present:integer
+1
+;
+
+TO_INTEGER(PRESENT) with filter returns 0
+required_capability: fn_present
+
+// tag::present-as-integer[]
+FROM employees
+| WHERE emp_no == 10020
+| STATS is_present = TO_INTEGER(PRESENT(languages))
+// end::present-as-integer[]
+;
+
+// tag::present-as-integer-result[]
+is_present:integer
+0
+// end::present-as-integer-result[]
+;

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

@@ -1472,7 +1472,12 @@ public class EsqlCapabilities {
         /**
          * TO_DENSE_VECTOR function.
          */
-        TO_DENSE_VECTOR_FUNCTION(Build.current().isSnapshot());
+        TO_DENSE_VECTOR_FUNCTION(Build.current().isSnapshot()),
+
+        /**
+         * Support present_over_time aggregation that gets evaluated per time-series
+         */
+        PRESENT_OVER_TIME(Build.current().isSnapshot());
 
         private final boolean enabled;
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java

@@ -63,6 +63,8 @@ import org.elasticsearch.xpack.esql.expression.function.aggregate.Max;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.MaxOverTime;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Min;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.MinOverTime;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.Present;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.PresentOverTime;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Sum;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.SumOverTime;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.SummationMode;
@@ -2129,6 +2131,9 @@ public class Analyzer extends ParameterizedRuleExecutor<LogicalPlan, AnalyzerCon
             if (aggFunc instanceof Avg || aggFunc instanceof AvgOverTime) {
                 return AggregateMetricDoubleBlockBuilder.Metric.COUNT;
             }
+            if (aggFunc instanceof Present || aggFunc instanceof PresentOverTime) {
+                return AggregateMetricDoubleBlockBuilder.Metric.COUNT;
+            }
             return null;
         }
     }

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

@@ -38,6 +38,7 @@ import org.elasticsearch.xpack.esql.expression.function.aggregate.Min;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.MinOverTime;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Percentile;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Present;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.PresentOverTime;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Rate;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Sample;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.SpatialCentroid;
@@ -529,7 +530,8 @@ public class EsqlFunctionRegistry {
                 def(Magnitude.class, Magnitude::new, "v_magnitude"),
                 def(Hamming.class, Hamming::new, "v_hamming"),
                 def(UrlEncode.class, UrlEncode::new, "url_encode"),
-                def(UrlDecode.class, UrlDecode::new, "url_decode") } };
+                def(UrlDecode.class, UrlDecode::new, "url_decode"),
+                def(PresentOverTime.class, uni(PresentOverTime::new), "present_over_time") } };
     }
 
     public EsqlFunctionRegistry snapshotRegistry() {

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

@@ -43,7 +43,8 @@ public class AggregateWritables {
             CountOverTime.ENTRY,
             CountDistinctOverTime.ENTRY,
             WeightedAvg.ENTRY,
-            Present.ENTRY
+            Present.ENTRY,
+            PresentOverTime.ENTRY
         );
     }
 }

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

@@ -48,6 +48,11 @@ public class Present extends AggregateFunction implements ToAggregator {
                 description = "To check for the presence inside a group use `PRESENT()` and `BY` clauses",
                 file = "present",
                 tag = "present-by"
+            ),
+            @Example(
+                description = "To check for the presence and return 1 when it's true and 0 when it's false",
+                file = "present",
+                tag = "present-as-integer"
             ) }
     )
     public Present(

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

@@ -0,0 +1,118 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.expression.function.aggregate;
+
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.expression.Literal;
+import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.expression.function.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 java.io.IOException;
+import java.util.List;
+
+import static java.util.Collections.emptyList;
+
+/**
+ * Similar to {@link Present}, but it is used to check the presence of values over a time series in the given field.
+ */
+public class PresentOverTime extends TimeSeriesAggregateFunction {
+    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
+        Expression.class,
+        "PresentOverTime",
+        PresentOverTime::new
+    );
+
+    @FunctionInfo(
+        type = FunctionType.TIME_SERIES_AGGREGATE,
+        returnType = { "boolean" },
+        description = "The presence of a field in the output result over time range.",
+        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 = "present_over_time") }
+    )
+    public PresentOverTime(
+        Source source,
+        @Param(
+            name = "field",
+            type = {
+                "aggregate_metric_double",
+                "boolean",
+                "cartesian_point",
+                "cartesian_shape",
+                "date",
+                "date_nanos",
+                "double",
+                "geo_point",
+                "geo_shape",
+                "geohash",
+                "geotile",
+                "geohex",
+                "integer",
+                "ip",
+                "keyword",
+                "long",
+                "text",
+                "unsigned_long",
+                "version" }
+        ) Expression field
+    ) {
+        this(source, field, Literal.TRUE);
+    }
+
+    public PresentOverTime(Source source, Expression field, Expression filter) {
+        super(source, field, filter, emptyList());
+    }
+
+    private PresentOverTime(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return ENTRY.name;
+    }
+
+    @Override
+    public PresentOverTime withFilter(Expression filter) {
+        return new PresentOverTime(source(), field(), filter);
+    }
+
+    @Override
+    protected NodeInfo<PresentOverTime> info() {
+        return NodeInfo.create(this, PresentOverTime::new, field(), filter());
+    }
+
+    @Override
+    public PresentOverTime replaceChildren(List<Expression> newChildren) {
+        return new PresentOverTime(source(), newChildren.get(0), newChildren.get(1));
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        return perTimeSeriesAggregation().resolveType();
+    }
+
+    @Override
+    public DataType dataType() {
+        return perTimeSeriesAggregation().dataType();
+    }
+
+    @Override
+    public Present perTimeSeriesAggregation() {
+        return new Present(source(), field(), filter());
+    }
+}

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

@@ -0,0 +1,35 @@
+/*
+ * 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.expression.function.AbstractFunctionTestCase;
+import org.elasticsearch.xpack.esql.expression.function.TestCaseSupplier;
+
+import java.util.List;
+import java.util.function.Supplier;
+
+public class PresentOverTimeTests extends AbstractFunctionTestCase {
+    public PresentOverTimeTests(@Name("TestCase") Supplier<TestCaseSupplier.TestCase> testCaseSupplier) {
+        this.testCase = testCaseSupplier.get();
+    }
+
+    @ParametersFactory
+    public static Iterable<Object[]> parameters() {
+        return PresentTests.parameters();
+    }
+
+    @Override
+    protected Expression build(Source source, List<Expression> args) {
+        return new PresentOverTime(source, args.get(0));
+    }
+}

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

@@ -42,6 +42,7 @@ setup:
             - count_over_time
             - count_distinct_over_time
             - cosine_vector_similarity_function
+            - present_over_time
       reason: "Test that should only be executed on snapshot versions"
 
   - do: {xpack.usage: {}}
@@ -129,7 +130,7 @@ setup:
   - match: {esql.functions.coalesce: $functions_coalesce}
   - gt: {esql.functions.categorize: $functions_categorize}
   # Testing for the entire function set isn't feasible, so we just check that we return the correct count as an approximation.
-  - length: {esql.functions: 175} # check the "sister" test below for a likely update to the same esql.functions length check
+  - length: {esql.functions: 176} # check the "sister" test below for a likely update to the same esql.functions length check
 ---
 "Basic ESQL usage output (telemetry) non-snapshot version":
   - requires: