Introduce `version` type support (ESQL-1218)

This adds support for the type `version`.
The conversion function `to_version()` has also beed added;
`to_string()` now supports the type as well.

docs/reference/esql/esql-functions.asciidoc

@@ -41,6 +41,7 @@ these functions:
 * <<esql-to_ip>>
 * <<esql-to_long>>
 * <<esql-to_string>>
+* <<esql-to_version>>
 
 include::functions/abs.asciidoc[]
 include::functions/auto_bucket.asciidoc[]
@@ -73,3 +74,4 @@ include::functions/to_integer.asciidoc[]
 include::functions/to_ip.asciidoc[]
 include::functions/to_long.asciidoc[]
 include::functions/to_string.asciidoc[]
+include::functions/to_version.asciidoc[]

docs/reference/esql/functions/to_version.asciidoc

@@ -0,0 +1,19 @@
+[[esql-to_version]]
+=== `TO_VERSION`
+Converts an input string to a version value. For example:
+
+[source,esql]
+----
+include::{esql-specs}/version.csv-spec[tag=to_version]
+----
+
+which returns:
+
+[%header,format=dsv,separator=|]
+|===
+include::{esql-specs}/version.csv-spec[tag=to_version-result]
+|===
+
+The input can be a single- or multi-valued field or an expression.
+
+Alias: TO_VER

x-pack/plugin/esql/qa/server/single-node/src/yamlRestTest/resources/rest-api-spec/test/30_types.yml

@@ -477,3 +477,31 @@ alias:
   - match: { columns.0.type: long }
   - length: { values: 1 }
   - match: { values.0.0: 50 }
+
+---
+version:
+  - do:
+      indices.create:
+        index: test
+        body:
+          mappings:
+            properties:
+              version:
+                type: version
+
+  - do:
+      bulk:
+        index: test
+        refresh: true
+        body:
+          - { "index": { } }
+          - { "version": [ "1.2.3", "4.5.6-SNOOPY" ] }
+
+  - do:
+      esql.query:
+        body:
+          query: 'from test'
+  - match: { columns.0.name: version }
+  - match: { columns.0.type: version }
+  - length: { values: 1 }
+  - match: { values.0.0: [ "1.2.3", "4.5.6-SNOOPY" ] }

x-pack/plugin/esql/qa/server/single-node/src/yamlRestTest/resources/rest-api-spec/test/40_unsupported_types.yml

@@ -12,8 +12,6 @@ unsupported:
                 type: aggregate_metric_double
                 metrics: [ min, max ]
                 default_metric: max
-              boolean:
-                type: boolean
               binary:
                 type: binary
               completion:
@@ -68,8 +66,6 @@ unsupported:
               token_count:
                 type: token_count
                 analyzer: standard
-              version:
-                type: version
 
   - do:
       bulk:
@@ -80,7 +76,6 @@ unsupported:
           - {
             "aggregate_metric_double": { "min": 1.0, "max": 3.0 },
             "binary": "U29tZSBiaW5hcnkgYmxvYg==",
-            "boolean": false,
             "completion": "foo bar",
             "date_nanos": "2015-01-01T12:10:30.123456789Z",
             "date_range": { "gte": "2015-10-31 12:00:00", "lte": "2050-12-31 12:00:00" },
@@ -100,8 +95,7 @@ unsupported:
             "shape": "LINESTRING (-377.03653 389.897676, -377.009051 389.889939)",
             "text": "foo bar",
             "token_count": "foo bar baz",
-            "some_doc": { "foo": "xy", "bar": 12 },
-            "version": "2.3.0"
+            "some_doc": { "foo": "xy", "bar": 12 }
           }
 
   - do:
@@ -112,65 +106,61 @@ unsupported:
   - match: { columns.0.type: unsupported }
   - match: { columns.1.name: binary }
   - match: { columns.1.type: unsupported }
-  - match: { columns.2.name: boolean }
-  - match: { columns.2.type: boolean }
-  - match: { columns.3.name: completion }
+  - match: { columns.2.name: completion }
+  - match: { columns.2.type: unsupported }
+  - match: { columns.3.name: date_nanos }
   - match: { columns.3.type: unsupported }
-  - match: { columns.4.name: date_nanos }
+  - match: { columns.4.name: date_range }
   - match: { columns.4.type: unsupported }
-  - match: { columns.5.name: date_range }
+  - match: { columns.5.name: dense_vector }
   - match: { columns.5.type: unsupported }
-  - match: { columns.6.name: dense_vector }
+  - match: { columns.6.name: double_range }
   - match: { columns.6.type: unsupported }
-  - match: { columns.7.name: double_range }
+  - match: { columns.7.name: float_range }
   - match: { columns.7.type: unsupported }
-  - match: { columns.8.name: float_range }
+  - match: { columns.8.name: geo_point }
   - match: { columns.8.type: unsupported }
-  - match: { columns.9.name: geo_point }
+  - match: { columns.9.name: geo_point_alias }
   - match: { columns.9.type: unsupported }
-  - match: { columns.10.name: geo_point_alias }
+  - match: { columns.10.name: histogram }
   - match: { columns.10.type: unsupported }
-  - match: { columns.11.name: histogram }
+  - match: { columns.11.name: integer_range }
   - match: { columns.11.type: unsupported }
-  - match: { columns.12.name: integer_range }
+  - match: { columns.12.name: ip_range }
   - match: { columns.12.type: unsupported }
-  - match: { columns.13.name: ip_range }
+  - match: { columns.13.name: long_range }
   - match: { columns.13.type: unsupported }
-  - match: { columns.14.name: long_range }
+  - match: { columns.14.name: match_only_text }
   - match: { columns.14.type: unsupported }
-  - match: { columns.15.name: match_only_text }
-  - match: { columns.15.type: unsupported }
-  - match: { columns.16.name: name }
-  - match: { columns.16.type: keyword }
-  - match: { columns.17.name: rank_feature }
+  - match: { columns.15.name: name }
+  - match: { columns.15.type: keyword }
+  - match: { columns.16.name: rank_feature }
+  - match: { columns.16.type: unsupported }
+  - match: { columns.17.name: rank_features }
   - match: { columns.17.type: unsupported }
-  - match: { columns.18.name: rank_features }
+  - match: { columns.18.name: search_as_you_type }
   - match: { columns.18.type: unsupported }
-  - match: { columns.19.name: search_as_you_type }
+  - match: { columns.19.name: search_as_you_type._2gram }
   - match: { columns.19.type: unsupported }
-  - match: { columns.20.name: search_as_you_type._2gram }
+  - match: { columns.20.name: search_as_you_type._3gram }
   - match: { columns.20.type: unsupported }
-  - match: { columns.21.name: search_as_you_type._3gram }
+  - match: { columns.21.name: search_as_you_type._index_prefix }
   - match: { columns.21.type: unsupported }
-  - match: { columns.22.name: search_as_you_type._index_prefix }
+  - match: { columns.22.name: shape }
   - match: { columns.22.type: unsupported }
-  - match: { columns.23.name: shape }
-  - match: { columns.23.type: unsupported }
-  - match: { columns.24.name: some_doc.bar }
-  - match: { columns.24.type: long }
-  - match: { columns.25.name: some_doc.foo }
-  - match: { columns.25.type: keyword }
-  - match: { columns.26.name: text }
-  - match: { columns.26.type: unsupported }
-  - match: { columns.27.name: token_count }
-  - match: { columns.27.type: integer }
-  - match: { columns.28.name: version }
-  - match: { columns.28.type: unsupported }
+  - match: { columns.23.name: some_doc.bar }
+  - match: { columns.23.type: long }
+  - match: { columns.24.name: some_doc.foo }
+  - match: { columns.24.type: keyword }
+  - match: { columns.25.name: text }
+  - match: { columns.25.type: unsupported }
+  - match: { columns.26.name: token_count }
+  - match: { columns.26.type: integer }
 
   - length: { values: 1 }
   - match: { values.0.0: "<unsupported>" }
   - match: { values.0.1: "<unsupported>" }
-  - match: { values.0.2: false }
+  - match: { values.0.2: "<unsupported>" }
   - match: { values.0.3: "<unsupported>" }
   - match: { values.0.4: "<unsupported>" }
   - match: { values.0.5: "<unsupported>" }
@@ -183,20 +173,18 @@ unsupported:
   - match: { values.0.12: "<unsupported>" }
   - match: { values.0.13: "<unsupported>" }
   - match: { values.0.14: "<unsupported>" }
-  - match: { values.0.15: "<unsupported>" }
-  - match: { values.0.16: Alice }
+  - match: { values.0.15: Alice }
+  - match: { values.0.16: "<unsupported>" }
   - match: { values.0.17: "<unsupported>" }
   - match: { values.0.18: "<unsupported>" }
   - match: { values.0.19: "<unsupported>" }
   - match: { values.0.20: "<unsupported>" }
   - match: { values.0.21: "<unsupported>" }
   - match: { values.0.22: "<unsupported>" }
-  - match: { values.0.23: "<unsupported>" }
-  - match: { values.0.24: 12 }
-  - match: { values.0.25: xy }
-  - match: { values.0.26: "<unsupported>" }
-  - match: { values.0.27: 3 }
-  - match: { values.0.28: "<unsupported>" }
+  - match: { values.0.23: 12 }
+  - match: { values.0.24: xy }
+  - match: { values.0.25: "<unsupported>" }
+  - match: { values.0.26: 3 }
 
 
 # limit 0
@@ -208,60 +196,56 @@ unsupported:
   - match: { columns.0.type: unsupported }
   - match: { columns.1.name: binary }
   - match: { columns.1.type: unsupported }
-  - match: { columns.2.name: boolean }
-  - match: { columns.2.type: boolean }
-  - match: { columns.3.name: completion }
+  - match: { columns.2.name: completion }
+  - match: { columns.2.type: unsupported }
+  - match: { columns.3.name: date_nanos }
   - match: { columns.3.type: unsupported }
-  - match: { columns.4.name: date_nanos }
+  - match: { columns.4.name: date_range }
   - match: { columns.4.type: unsupported }
-  - match: { columns.5.name: date_range }
+  - match: { columns.5.name: dense_vector }
   - match: { columns.5.type: unsupported }
-  - match: { columns.6.name: dense_vector }
+  - match: { columns.6.name: double_range }
   - match: { columns.6.type: unsupported }
-  - match: { columns.7.name: double_range }
+  - match: { columns.7.name: float_range }
   - match: { columns.7.type: unsupported }
-  - match: { columns.8.name: float_range }
+  - match: { columns.8.name: geo_point }
   - match: { columns.8.type: unsupported }
-  - match: { columns.9.name: geo_point }
+  - match: { columns.9.name: geo_point_alias }
   - match: { columns.9.type: unsupported }
-  - match: { columns.10.name: geo_point_alias }
+  - match: { columns.10.name: histogram }
   - match: { columns.10.type: unsupported }
-  - match: { columns.11.name: histogram }
+  - match: { columns.11.name: integer_range }
   - match: { columns.11.type: unsupported }
-  - match: { columns.12.name: integer_range }
+  - match: { columns.12.name: ip_range }
   - match: { columns.12.type: unsupported }
-  - match: { columns.13.name: ip_range }
+  - match: { columns.13.name: long_range }
   - match: { columns.13.type: unsupported }
-  - match: { columns.14.name: long_range }
+  - match: { columns.14.name: match_only_text }
   - match: { columns.14.type: unsupported }
-  - match: { columns.15.name: match_only_text }
-  - match: { columns.15.type: unsupported }
-  - match: { columns.16.name: name }
-  - match: { columns.16.type: keyword }
-  - match: { columns.17.name: rank_feature }
+  - match: { columns.15.name: name }
+  - match: { columns.15.type: keyword }
+  - match: { columns.16.name: rank_feature }
+  - match: { columns.16.type: unsupported }
+  - match: { columns.17.name: rank_features }
   - match: { columns.17.type: unsupported }
-  - match: { columns.18.name: rank_features }
+  - match: { columns.18.name: search_as_you_type }
   - match: { columns.18.type: unsupported }
-  - match: { columns.19.name: search_as_you_type }
+  - match: { columns.19.name: search_as_you_type._2gram }
   - match: { columns.19.type: unsupported }
-  - match: { columns.20.name: search_as_you_type._2gram }
+  - match: { columns.20.name: search_as_you_type._3gram }
   - match: { columns.20.type: unsupported }
-  - match: { columns.21.name: search_as_you_type._3gram }
+  - match: { columns.21.name: search_as_you_type._index_prefix }
   - match: { columns.21.type: unsupported }
-  - match: { columns.22.name: search_as_you_type._index_prefix }
+  - match: { columns.22.name: shape }
   - match: { columns.22.type: unsupported }
-  - match: { columns.23.name: shape }
-  - match: { columns.23.type: unsupported }
-  - match: { columns.24.name: some_doc.bar }
-  - match: { columns.24.type: long }
-  - match: { columns.25.name: some_doc.foo }
-  - match: { columns.25.type: keyword }
-  - match: { columns.26.name: text }
-  - match: { columns.26.type: unsupported }
-  - match: { columns.27.name: token_count }
-  - match: { columns.27.type: integer }
-  - match: { columns.28.name: version }
-  - match: { columns.28.type: unsupported }
+  - match: { columns.23.name: some_doc.bar }
+  - match: { columns.23.type: long }
+  - match: { columns.24.name: some_doc.foo }
+  - match: { columns.24.type: keyword }
+  - match: { columns.25.name: text }
+  - match: { columns.25.type: unsupported }
+  - match: { columns.26.name: token_count }
+  - match: { columns.26.type: integer }
 
   - length: { values: 0 }
 

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

@@ -12,6 +12,7 @@ import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.logging.Logger;
 import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.xpack.esql.CsvTestUtils.ActualResults;
+import org.elasticsearch.xpack.versionfield.Version;
 import org.hamcrest.Matchers;
 
 import java.util.ArrayList;
@@ -117,7 +118,7 @@ public final class CsvAssert {
                 if (blockType == Type.LONG && expectedType == Type.DATETIME) {
                     continue;
                 }
-                if (blockType == Type.KEYWORD && expectedType == Type.IP) {
+                if (blockType == Type.KEYWORD && (expectedType == Type.IP || expectedType == Type.VERSION)) {
                     // Type.asType translates all bytes references into keywords
                     continue;
                 }
@@ -180,6 +181,9 @@ public final class CsvAssert {
                         } else if (expectedType == Type.IP) {
                             // convert BytesRef-packed IP to String, allowing subsequent comparison with what's expected
                             expectedValue = rebuildExpected(expectedValue, BytesRef.class, x -> DocValueFormat.IP.format((BytesRef) x));
+                        } else if (expectedType == Type.VERSION) {
+                            // convert BytesRef-packed Version to String
+                            expectedValue = rebuildExpected(expectedValue, BytesRef.class, x -> new Version((BytesRef) x).toString());
                         }
 
                     }

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

@@ -21,6 +21,7 @@ import org.elasticsearch.core.Tuple;
 import org.elasticsearch.logging.Logger;
 import org.elasticsearch.xpack.esql.action.EsqlQueryResponse;
 import org.elasticsearch.xpack.ql.util.StringUtils;
+import org.elasticsearch.xpack.versionfield.Version;
 import org.supercsv.io.CsvListReader;
 import org.supercsv.prefs.CsvPreference;
 
@@ -245,6 +246,9 @@ public final class CsvTestUtils {
                 String name = nameWithType[0].trim();
                 columnNames.add(name);
                 Type type = Type.asType(typeName);
+                if (type == null) {
+                    throw new IllegalArgumentException("Unknown type name: [" + typeName + "]");
+                }
                 columnTypes.add(type);
             }
 
@@ -301,6 +305,7 @@ public final class CsvTestUtils {
         SCALED_FLOAT(s -> s == null ? null : scaledFloat(s, "100"), Double.class),
         KEYWORD(Object::toString, BytesRef.class),
         IP(StringUtils::parseIP, BytesRef.class),
+        VERSION(v -> new Version(v).toBytesRef(), BytesRef.class),
         NULL(s -> null, Void.class),
         DATETIME(x -> x == null ? null : DateFormatters.from(UTC_DATE_TIME_FORMATTER.parse(x)).toInstant().toEpochMilli(), Long.class),
         BOOLEAN(Booleans::parseBoolean, Boolean.class);
@@ -325,6 +330,7 @@ public final class CsvTestUtils {
             LOOKUP.put("N", NULL);
             LOOKUP.put("DATE", DATETIME);
             LOOKUP.put("DT", DATETIME);
+            LOOKUP.put("V", VERSION);
         }
 
         private final Function<String, Object> converter;

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

@@ -48,7 +48,15 @@ import static org.elasticsearch.xpack.esql.CsvTestUtils.multiValuesAwareCsvToStr
 public class CsvTestsDataLoader {
     private static final TestsDataset EMPLOYEES = new TestsDataset("employees", "mapping-default.json", "employees.csv");
     private static final TestsDataset HOSTS = new TestsDataset("hosts", "mapping-hosts.json", "hosts.csv");
-    public static final Map<String, TestsDataset> CSV_DATASET_MAP = Map.of(EMPLOYEES.indexName, EMPLOYEES, HOSTS.indexName, HOSTS);
+    private static final TestsDataset APPS = new TestsDataset("apps", "mapping-apps.json", "apps.csv");
+    public static final Map<String, TestsDataset> CSV_DATASET_MAP = Map.of(
+        EMPLOYEES.indexName,
+        EMPLOYEES,
+        HOSTS.indexName,
+        HOSTS,
+        APPS.indexName,
+        APPS
+    );
 
     /**
      * <p>

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

@@ -0,0 +1,15 @@
+id:integer,version:version,name:keyword
+1,1,aaaaa
+2,2.1,bbbbb
+3,2.3.4,ccccc
+4,2.12.0,ddddd
+5,1.11.0,eeeee
+6,5.2.9,fffff
+7,5.2.9-SNAPSHOT,ggggg
+8,1.2.3.4,hhhhh
+9,bad,iiiii
+10,5.2.9,jjjjj
+11,,kkkkk
+12,1.2.3.4,aaaaa
+13,,lllll
+14,5.2.9,mmmmm

x-pack/plugin/esql/qa/testFixtures/src/main/resources/mapping-apps.json

@@ -0,0 +1,13 @@
+{
+    "properties" : {
+        "id" : {
+            "type" : "integer"
+        },
+        "version" : {
+            "type" : "version"
+        },
+        "name" : {
+            "type" : "keyword"
+        }
+    }
+}

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

@@ -54,6 +54,8 @@ to_ip                    |to_ip(arg1)
 to_long                  |to_long(arg1)                  
 to_str                   |to_str(arg1)                   
 to_string                |to_string(arg1) 
+to_ver                   |to_ver(arg1) 
+to_version               |to_version(arg1) 
 ;
 
 showFunctionsFiltered

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

@@ -0,0 +1,309 @@
+// To mute tests follow example in file: example.csv-spec
+
+//
+// Tests for VERSION fields
+//
+
+selectAll
+FROM apps;
+
+id:integer |name:keyword   |version:version
+1          |aaaaa          |1
+2          |bbbbb          |2.1
+3          |ccccc          |2.3.4
+4          |ddddd          |2.12.0
+5          |eeeee          |1.11.0
+6          |fffff          |5.2.9
+7          |ggggg          |5.2.9-SNAPSHOT
+8          |hhhhh          |1.2.3.4
+9          |iiiii          |bad
+10         |jjjjj          |5.2.9
+11         |kkkkk          |null
+12         |aaaaa          |1.2.3.4
+13         |lllll          |null
+14         |mmmmm          |5.2.9
+;
+
+filterByVersion
+FROM apps | WHERE version == to_ver("2.12.0");
+
+id:i |name:k |version:v
+4    |ddddd  |2.12.0
+;
+
+projectionVersion
+FROM apps | WHERE id == 3 | PROJECT version;
+
+version:v
+2.3.4
+;
+
+versionRange1
+FROM apps | WHERE version > to_ver("2.2") | SORT id;
+
+id:i  |name:k     |version:v
+3     |ccccc      |2.3.4
+4     |ddddd      |2.12.0
+6     |fffff      |5.2.9
+7     |ggggg      |5.2.9-SNAPSHOT
+9     |iiiii      |bad
+10    |jjjjj      |5.2.9
+14    |mmmmm      |5.2.9
+;
+
+versionRange2
+FROM apps | WHERE version >= to_ver("2.3.4") | SORT id;
+
+id:i  |name:k     |version:v
+3     |ccccc      |2.3.4
+4     |ddddd      |2.12.0
+6     |fffff      |5.2.9
+7     |ggggg      |5.2.9-SNAPSHOT
+9     |iiiii      |bad
+10    |jjjjj      |5.2.9
+14    |mmmmm      |5.2.9
+;
+
+between
+FROM apps | WHERE version >= to_ver("1.10") AND version <= to_ver("5.2.9") | SORT id;
+
+id:i    |name:k       |version:v
+2       |bbbbb        | 2.1
+3       |ccccc        | 2.3.4
+4       |ddddd        | 2.12.0
+5       |eeeee        | 1.11.0
+6       |fffff        | 5.2.9
+7       |ggggg        | 5.2.9-SNAPSHOT
+10      |jjjjj        | 5.2.9
+14      |mmmmm        | 5.2.9
+;
+
+orderByVersion
+FROM apps | SORT version, id;
+
+id:i  |name:s     |version:v
+1     |aaaaa      |1
+8     |hhhhh      |1.2.3.4
+12    |aaaaa      |1.2.3.4
+5     |eeeee      |1.11.0
+2     |bbbbb      |2.1
+3     |ccccc      |2.3.4
+4     |ddddd      |2.12.0
+7     |ggggg      |5.2.9-SNAPSHOT
+6     |fffff      |5.2.9
+10    |jjjjj      |5.2.9
+14    |mmmmm      |5.2.9
+9     |iiiii      |bad
+11    |kkkkk      |null
+13    |lllll      |null
+;
+
+orderByVersionDesc
+FROM apps | SORT version DESC, id ASC;
+
+id:i  |name:s     |version:v
+11    |kkkkk      |null
+13    |lllll      |null
+9     |iiiii      |bad
+6     |fffff      |5.2.9
+10    |jjjjj      |5.2.9
+14    |mmmmm      |5.2.9
+7     |ggggg      |5.2.9-SNAPSHOT
+4     |ddddd      |2.12.0
+3     |ccccc      |2.3.4
+2     |bbbbb      |2.1
+5     |eeeee      |1.11.0
+8     |hhhhh      |1.2.3.4
+12    |aaaaa      |1.2.3.4
+1     |aaaaa      |1
+;
+
+orderByVersionNullsFirst
+FROM apps | SORT version NULLS FIRST, id;
+
+id:i  |name:s     |version:v
+11    |kkkkk      |null
+13    |lllll      |null
+1     |aaaaa      |1
+8     |hhhhh      |1.2.3.4
+12    |aaaaa      |1.2.3.4
+5     |eeeee      |1.11.0
+2     |bbbbb      |2.1
+3     |ccccc      |2.3.4
+4     |ddddd      |2.12.0
+7     |ggggg      |5.2.9-SNAPSHOT
+6     |fffff      |5.2.9
+10    |jjjjj      |5.2.9
+14    |mmmmm      |5.2.9
+9     |iiiii      |bad
+;
+
+orderByVersionMultipleCasts
+FROM apps | EVAL o = TO_VER(CONCAT("1.", TO_STR(version))) | SORT o, id;
+
+id:i           |name:s         |version:v      |o:v        
+1              |aaaaa          |1              |1.1             
+8              |hhhhh          |1.2.3.4        |1.1.2.3.4       
+12             |aaaaa          |1.2.3.4        |1.1.2.3.4       
+5              |eeeee          |1.11.0         |1.1.11.0        
+2              |bbbbb          |2.1            |1.2.1           
+3              |ccccc          |2.3.4          |1.2.3.4         
+4              |ddddd          |2.12.0         |1.2.12.0        
+7              |ggggg          |5.2.9-SNAPSHOT |1.5.2.9-SNAPSHOT
+6              |fffff          |5.2.9          |1.5.2.9         
+10             |jjjjj          |5.2.9          |1.5.2.9         
+14             |mmmmm          |5.2.9          |1.5.2.9         
+9              |iiiii          |bad            |1.bad           
+11             |kkkkk          |null           |null            
+13             |lllll          |null           |null 
+;
+
+countVersion
+FROM apps | RENAME k = name | STATS v = COUNT(version) BY k | SORT k;
+
+v:l     | k:s
+2       | aaaaa
+1       | bbbbb
+1       | ccccc
+1       | ddddd
+1       | eeeee
+1       | fffff
+1       | ggggg
+1       | hhhhh
+1       | iiiii
+1       | jjjjj
+0       | kkkkk
+0       | lllll
+1       | mmmmm
+;
+
+groupByVersion
+FROM apps | STATS c = COUNT(version), maxid = MAX(id) BY version | SORT version;
+
+c:l |maxid:i  |version:v
+// 2   |13       |null # https://github.com/elastic/elasticsearch-internal/issues/770
+1   |1        |1
+2   |12       |1.2.3.4
+1   |5        |1.11.0
+1   |2        |2.1
+1   |3        |2.3.4
+1   |4        |2.12.0
+1   |7        |5.2.9-SNAPSHOT
+3   |14       |5.2.9
+1   |9        |bad
+;
+
+groupOrderLimit
+FROM apps | WHERE not is_null(version) | STATS c = COUNT(version) BY version | SORT version DESC | DROP c | LIMIT 3;
+
+version:v
+bad
+5.2.9
+5.2.9-SNAPSHOT
+;
+
+groupByVersionCast
+FROM apps | EVAL g = TO_VER(CONCAT("1.", TO_STR(version))) | STATS id = MAX(id) BY g | SORT id | DROP g;
+
+id:i
+1
+2
+3
+4
+5
+7
+9
+12
+// 13 # https://github.com/elastic/elasticsearch-internal/issues/770
+14
+;
+
+castConstantToVersion
+// tag::to_version[]
+ROW v = TO_VERSION("1.2.3")
+// end::to_version[]
+;
+
+// tag::to_version-result[]
+v:version
+1.2.3
+// end::to_version-result[]
+;
+
+castConstantToVersion2
+FROM apps | EVAL v = TO_VERSION("1.2.3") | PROJECT v;
+
+v:v
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+;
+
+multipleCast
+FROM apps | EVAL v = TO_STR(TO_VER("1.2.3")) | PROJECT v;
+
+v:s
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+1.2.3
+;
+
+compareVersions
+ROW v1 = TO_VER("1.2.3"), v2 = TO_VER("1.11.4") | EVAL v = v1 < v2 | PROJECT v;
+
+v:boolean
+true
+;
+
+groupByVersionAfterStats
+FROM apps | STATS idx = MAX(id) BY version | WHERE idx == 14;
+
+idx:i  |version:v
+14     | 5.2.9
+;
+
+case
+FROM apps
+| EVAL version_text = TO_STR(version)
+| WHERE IS_NULL(version) OR version_text LIKE "1*"
+| EVAL v = TO_VER(CONCAT("123", TO_STR(version)))
+| EVAL m = CASE(version > TO_VER("1.1"), 1, 0)
+| EVAL g = CASE(version > TO_VER("1.3.0"), version, TO_VER("1.3.0"))
+| EVAL i = CASE(IS_NULL(version), TO_VER("0.1"), version)
+| EVAL c = CASE(
+    version > TO_VER("1.1"), "high",
+    IS_NULL(version), "none",
+    "low")
+| SORT version DESC NULLS LAST, id DESC
+| PROJECT v, version, version_text, id, m, g, i, c;
+
+v:v        | version:v |version_text:s   | id:i  |  m:i  |  g:v   | i:v     |  c:s
+1231.11.0  | 1.11.0    | 1.11.0          | 5     | 1     | 1.11.0 | 1.11.0  | high
+1231.2.3.4 | 1.2.3.4   | 1.2.3.4         | 12    | 1     | 1.3.0  | 1.2.3.4 | high
+1231.2.3.4 | 1.2.3.4   | 1.2.3.4         | 8     | 1     | 1.3.0  | 1.2.3.4 | high
+1231       | 1         | 1               | 1     | 0     | 1.3.0  | 1       | low
+null       | null      | null            | 13    | 0     | 1.3.0  | 0.1     | none
+null       | null      | null            | 11    | 0     | 1.3.0  | 0.1     | none
+;

x-pack/plugin/esql/src/main/java/generated/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToStringFromVersionEvaluator.java

@@ -0,0 +1,112 @@
+// 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.scalar.convert;
+
+import java.lang.Override;
+import java.lang.String;
+import java.util.BitSet;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.BytesRefArray;
+import org.elasticsearch.compute.data.Block;
+import org.elasticsearch.compute.data.BytesRefArrayBlock;
+import org.elasticsearch.compute.data.BytesRefArrayVector;
+import org.elasticsearch.compute.data.BytesRefBlock;
+import org.elasticsearch.compute.data.BytesRefVector;
+import org.elasticsearch.compute.data.ConstantBytesRefVector;
+import org.elasticsearch.compute.data.Vector;
+import org.elasticsearch.compute.operator.EvalOperator;
+import org.elasticsearch.xpack.ql.tree.Source;
+
+/**
+ * {@link EvalOperator.ExpressionEvaluator} implementation for {@link ToString}.
+ * This class is generated. Do not edit it.
+ */
+public final class ToStringFromVersionEvaluator extends AbstractConvertFunction.AbstractEvaluator {
+  public ToStringFromVersionEvaluator(EvalOperator.ExpressionEvaluator field, Source source) {
+    super(field, source);
+  }
+
+  @Override
+  public String name() {
+    return "ToString";
+  }
+
+  @Override
+  public Block evalVector(Vector v) {
+    BytesRefVector vector = (BytesRefVector) v;
+    int positionCount = v.getPositionCount();
+    BytesRef scratchPad = new BytesRef();
+    if (vector.isConstant()) {
+      try {
+        return new ConstantBytesRefVector(evalValue(vector, 0, scratchPad), positionCount).asBlock();
+      } catch (Exception e) {
+        registerException(e);
+        return Block.constantNullBlock(positionCount);
+      }
+    }
+    BitSet nullsMask = null;
+    BytesRefArray values = new BytesRefArray(positionCount, BigArrays.NON_RECYCLING_INSTANCE);
+    for (int p = 0; p < positionCount; p++) {
+      try {
+        values.append(evalValue(vector, p, scratchPad));
+      } catch (Exception e) {
+        registerException(e);
+        if (nullsMask == null) {
+          nullsMask = new BitSet(positionCount);
+        }
+        nullsMask.set(p);
+      }
+    }
+    return nullsMask == null
+          ? new BytesRefArrayVector(values, positionCount).asBlock()
+          // UNORDERED, since whatever ordering there is, it isn't necessarily preserved
+          : new BytesRefArrayBlock(values, positionCount, null, nullsMask, Block.MvOrdering.UNORDERED);
+  }
+
+  private static BytesRef evalValue(BytesRefVector container, int index, BytesRef scratchPad) {
+    BytesRef value = container.getBytesRef(index, scratchPad);
+    return ToString.fromVersion(value);
+  }
+
+  @Override
+  public Block evalBlock(Block b) {
+    BytesRefBlock block = (BytesRefBlock) b;
+    int positionCount = block.getPositionCount();
+    BytesRefBlock.Builder builder = BytesRefBlock.newBlockBuilder(positionCount);
+    BytesRef scratchPad = new BytesRef();
+    for (int p = 0; p < positionCount; p++) {
+      int valueCount = block.getValueCount(p);
+      int start = block.getFirstValueIndex(p);
+      int end = start + valueCount;
+      boolean positionOpened = false;
+      boolean valuesAppended = false;
+      for (int i = start; i < end; i++) {
+        try {
+          BytesRef value = evalValue(block, i, scratchPad);
+          if (positionOpened == false && valueCount > 1) {
+            builder.beginPositionEntry();
+            positionOpened = true;
+          }
+          builder.appendBytesRef(value);
+          valuesAppended = true;
+        } catch (Exception e) {
+          registerException(e);
+        }
+      }
+      if (valuesAppended == false) {
+        builder.appendNull();
+      } else if (positionOpened) {
+        builder.endPositionEntry();
+      }
+    }
+    return builder.build();
+  }
+
+  private static BytesRef evalValue(BytesRefBlock container, int index, BytesRef scratchPad) {
+    BytesRef value = container.getBytesRef(index, scratchPad);
+    return ToString.fromVersion(value);
+  }
+}

x-pack/plugin/esql/src/main/java/generated/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToVersionFromStringEvaluator.java

@@ -0,0 +1,112 @@
+// 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.scalar.convert;
+
+import java.lang.Override;
+import java.lang.String;
+import java.util.BitSet;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.BytesRefArray;
+import org.elasticsearch.compute.data.Block;
+import org.elasticsearch.compute.data.BytesRefArrayBlock;
+import org.elasticsearch.compute.data.BytesRefArrayVector;
+import org.elasticsearch.compute.data.BytesRefBlock;
+import org.elasticsearch.compute.data.BytesRefVector;
+import org.elasticsearch.compute.data.ConstantBytesRefVector;
+import org.elasticsearch.compute.data.Vector;
+import org.elasticsearch.compute.operator.EvalOperator;
+import org.elasticsearch.xpack.ql.tree.Source;
+
+/**
+ * {@link EvalOperator.ExpressionEvaluator} implementation for {@link ToVersion}.
+ * This class is generated. Do not edit it.
+ */
+public final class ToVersionFromStringEvaluator extends AbstractConvertFunction.AbstractEvaluator {
+  public ToVersionFromStringEvaluator(EvalOperator.ExpressionEvaluator field, Source source) {
+    super(field, source);
+  }
+
+  @Override
+  public String name() {
+    return "ToVersion";
+  }
+
+  @Override
+  public Block evalVector(Vector v) {
+    BytesRefVector vector = (BytesRefVector) v;
+    int positionCount = v.getPositionCount();
+    BytesRef scratchPad = new BytesRef();
+    if (vector.isConstant()) {
+      try {
+        return new ConstantBytesRefVector(evalValue(vector, 0, scratchPad), positionCount).asBlock();
+      } catch (Exception e) {
+        registerException(e);
+        return Block.constantNullBlock(positionCount);
+      }
+    }
+    BitSet nullsMask = null;
+    BytesRefArray values = new BytesRefArray(positionCount, BigArrays.NON_RECYCLING_INSTANCE);
+    for (int p = 0; p < positionCount; p++) {
+      try {
+        values.append(evalValue(vector, p, scratchPad));
+      } catch (Exception e) {
+        registerException(e);
+        if (nullsMask == null) {
+          nullsMask = new BitSet(positionCount);
+        }
+        nullsMask.set(p);
+      }
+    }
+    return nullsMask == null
+          ? new BytesRefArrayVector(values, positionCount).asBlock()
+          // UNORDERED, since whatever ordering there is, it isn't necessarily preserved
+          : new BytesRefArrayBlock(values, positionCount, null, nullsMask, Block.MvOrdering.UNORDERED);
+  }
+
+  private static BytesRef evalValue(BytesRefVector container, int index, BytesRef scratchPad) {
+    BytesRef value = container.getBytesRef(index, scratchPad);
+    return ToVersion.fromKeyword(value);
+  }
+
+  @Override
+  public Block evalBlock(Block b) {
+    BytesRefBlock block = (BytesRefBlock) b;
+    int positionCount = block.getPositionCount();
+    BytesRefBlock.Builder builder = BytesRefBlock.newBlockBuilder(positionCount);
+    BytesRef scratchPad = new BytesRef();
+    for (int p = 0; p < positionCount; p++) {
+      int valueCount = block.getValueCount(p);
+      int start = block.getFirstValueIndex(p);
+      int end = start + valueCount;
+      boolean positionOpened = false;
+      boolean valuesAppended = false;
+      for (int i = start; i < end; i++) {
+        try {
+          BytesRef value = evalValue(block, i, scratchPad);
+          if (positionOpened == false && valueCount > 1) {
+            builder.beginPositionEntry();
+            positionOpened = true;
+          }
+          builder.appendBytesRef(value);
+          valuesAppended = true;
+        } catch (Exception e) {
+          registerException(e);
+        }
+      }
+      if (valuesAppended == false) {
+        builder.appendNull();
+      } else if (positionOpened) {
+        builder.endPositionEntry();
+      }
+    }
+    return builder.build();
+  }
+
+  private static BytesRef evalValue(BytesRefBlock container, int index, BytesRef scratchPad) {
+    BytesRef value = container.getBytesRef(index, scratchPad);
+    return ToVersion.fromKeyword(value);
+  }
+}

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

@@ -25,6 +25,7 @@ import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentType;
+import org.elasticsearch.xpack.versionfield.Version;
 
 import java.io.IOException;
 
@@ -154,6 +155,14 @@ public record ColumnInfo(String name, String type) implements Writeable {
                     return builder.value(((BooleanBlock) block).getBoolean(valueIndex));
                 }
             };
+            case "version" -> new PositionToXContent(block) {
+                @Override
+                protected XContentBuilder valueToXContent(XContentBuilder builder, ToXContent.Params params, int valueIndex)
+                    throws IOException {
+                    BytesRef val = ((BytesRefBlock) block).getBytesRef(valueIndex, scratch);
+                    return builder.value(new Version(val).toString());
+                }
+            };
             case "null" -> new PositionToXContent(block) {
                 @Override
                 protected XContentBuilder valueToXContent(XContentBuilder builder, ToXContent.Params params, int valueIndex)

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

@@ -31,6 +31,7 @@ import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner;
 import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
+import org.elasticsearch.xpack.versionfield.Version;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -231,6 +232,7 @@ public class EsqlQueryResponse extends ActionResponse implements ChunkedToXConte
                 yield UTC_DATE_TIME_FORMATTER.formatMillis(longVal);
             }
             case "boolean" -> ((BooleanBlock) block).getBoolean(offset);
+            case "version" -> new Version(((BytesRefBlock) block).getBytesRef(offset, scratch)).toString();
             case "unsupported" -> UnsupportedValueSource.UNSUPPORTED_OUTPUT;
             default -> throw new UnsupportedOperationException("unsupported data type [" + dataType + "]");
         };
@@ -261,6 +263,7 @@ public class EsqlQueryResponse extends ActionResponse implements ChunkedToXConte
                     }
                     case "boolean" -> ((BooleanBlock.Builder) builder).appendBoolean(((Boolean) value));
                     case "null" -> builder.appendNull();
+                    case "version" -> ((BytesRefBlock.Builder) builder).appendBytesRef(new Version(value.toString()).toBytesRef());
                     default -> throw new UnsupportedOperationException("unsupported data type [" + dataTypes.get(c) + "]");
                 }
             }

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

@@ -220,15 +220,16 @@ public class Verifier {
         allowed.add(DataTypes.KEYWORD);
         allowed.add(DataTypes.IP);
         allowed.add(DataTypes.DATETIME);
+        allowed.add(DataTypes.VERSION);
         if (bc instanceof Equals || bc instanceof NotEquals) {
             allowed.add(DataTypes.BOOLEAN);
         }
         Expression.TypeResolution r = TypeResolutions.isType(
             bc.left(),
-            t -> allowed.contains(t),
+            allowed::contains,
             bc.sourceText(),
             FIRST,
-            Stream.concat(Stream.of("numeric"), allowed.stream().map(a -> a.typeName())).toArray(String[]::new)
+            Stream.concat(Stream.of("numeric"), allowed.stream().map(DataType::typeName)).toArray(String[]::new)
         );
         if (false == r.resolved()) {
             return fail(bc, r.message());

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

@@ -24,6 +24,7 @@ import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToIP;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToInteger;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToLong;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToString;
+import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToVersion;
 import org.elasticsearch.xpack.esql.expression.function.scalar.date.DateFormat;
 import org.elasticsearch.xpack.esql.expression.function.scalar.date.DateParse;
 import org.elasticsearch.xpack.esql.expression.function.scalar.date.DateTrunc;
@@ -107,7 +108,8 @@ public class EsqlFunctionRegistry extends FunctionRegistry {
                 def(ToIP.class, ToIP::new, "to_ip"),
                 def(ToInteger.class, ToInteger::new, "to_integer", "to_int"),
                 def(ToLong.class, ToLong::new, "to_long"),
-                def(ToString.class, ToString::new, "to_string", "to_str"), },
+                def(ToString.class, ToString::new, "to_string", "to_str"),
+                def(ToVersion.class, ToVersion::new, "to_version", "to_ver"), },
             // multivalue functions
             new FunctionDefinition[] {
                 def(MvAvg.class, MvAvg::new, "mv_avg"),

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToString.java

@@ -16,6 +16,7 @@ import org.elasticsearch.xpack.ql.expression.Expression;
 import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.ql.type.DataType;
+import org.elasticsearch.xpack.versionfield.Version;
 
 import java.util.List;
 import java.util.Map;
@@ -28,6 +29,7 @@ import static org.elasticsearch.xpack.ql.type.DataTypes.INTEGER;
 import static org.elasticsearch.xpack.ql.type.DataTypes.IP;
 import static org.elasticsearch.xpack.ql.type.DataTypes.KEYWORD;
 import static org.elasticsearch.xpack.ql.type.DataTypes.LONG;
+import static org.elasticsearch.xpack.ql.type.DataTypes.VERSION;
 import static org.elasticsearch.xpack.ql.util.DateUtils.UTC_DATE_TIME_FORMATTER;
 
 public class ToString extends AbstractConvertFunction implements Mappable {
@@ -47,7 +49,9 @@ public class ToString extends AbstractConvertFunction implements Mappable {
             LONG,
             ToStringFromLongEvaluator::new,
             INTEGER,
-            ToStringFromIntEvaluator::new
+            ToStringFromIntEvaluator::new,
+            VERSION,
+            ToStringFromVersionEvaluator::new
         );
 
     public ToString(Source source, Expression field) {
@@ -103,4 +107,9 @@ public class ToString extends AbstractConvertFunction implements Mappable {
     static BytesRef fromDouble(int integer) {
         return new BytesRef(String.valueOf(integer));
     }
+
+    @ConvertEvaluator(extraName = "FromVersion")
+    static BytesRef fromVersion(BytesRef version) {
+        return new BytesRef(new Version(version).toString());
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToVersion.java

@@ -0,0 +1,59 @@
+/*
+ * 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.scalar.convert;
+
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.compute.ann.ConvertEvaluator;
+import org.elasticsearch.compute.operator.EvalOperator;
+import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.tree.NodeInfo;
+import org.elasticsearch.xpack.ql.tree.Source;
+import org.elasticsearch.xpack.ql.type.DataType;
+import org.elasticsearch.xpack.versionfield.Version;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiFunction;
+
+import static org.elasticsearch.xpack.ql.type.DataTypes.KEYWORD;
+import static org.elasticsearch.xpack.ql.type.DataTypes.VERSION;
+
+public class ToVersion extends AbstractConvertFunction {
+
+    private static final Map<DataType, BiFunction<EvalOperator.ExpressionEvaluator, Source, EvalOperator.ExpressionEvaluator>> EVALUATORS =
+        Map.of(VERSION, (fieldEval, source) -> fieldEval, KEYWORD, ToVersionFromStringEvaluator::new);
+
+    public ToVersion(Source source, Expression field) {
+        super(source, field);
+    }
+
+    @Override
+    protected Map<DataType, BiFunction<EvalOperator.ExpressionEvaluator, Source, EvalOperator.ExpressionEvaluator>> evaluators() {
+        return EVALUATORS;
+    }
+
+    @Override
+    public DataType dataType() {
+        return VERSION;
+    }
+
+    @Override
+    public Expression replaceChildren(List<Expression> newChildren) {
+        return new ToVersion(source(), newChildren.get(0));
+    }
+
+    @Override
+    protected NodeInfo<? extends Expression> info() {
+        return NodeInfo.create(this, ToVersion::new, field());
+    }
+
+    @ConvertEvaluator(extraName = "FromString")
+    static BytesRef fromKeyword(BytesRef asString) {
+        return new Version(asString.utf8ToString()).toBytesRef();
+    }
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/io/stream/PlanNamedTypes.java

@@ -32,6 +32,7 @@ import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToIP;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToInteger;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToLong;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToString;
+import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToVersion;
 import org.elasticsearch.xpack.esql.expression.function.scalar.date.DateFormat;
 import org.elasticsearch.xpack.esql.expression.function.scalar.date.DateParse;
 import org.elasticsearch.xpack.esql.expression.function.scalar.date.DateTrunc;
@@ -258,6 +259,7 @@ public final class PlanNamedTypes {
             of(ESQL_UNARY_SCLR_CLS, ToInteger.class, PlanNamedTypes::writeESQLUnaryScalar, PlanNamedTypes::readESQLUnaryScalar),
             of(ESQL_UNARY_SCLR_CLS, ToLong.class, PlanNamedTypes::writeESQLUnaryScalar, PlanNamedTypes::readESQLUnaryScalar),
             of(ESQL_UNARY_SCLR_CLS, ToString.class, PlanNamedTypes::writeESQLUnaryScalar, PlanNamedTypes::readESQLUnaryScalar),
+            of(ESQL_UNARY_SCLR_CLS, ToVersion.class, PlanNamedTypes::writeESQLUnaryScalar, PlanNamedTypes::readESQLUnaryScalar),
             // ScalarFunction
             of(ScalarFunction.class, AutoBucket.class, PlanNamedTypes::writeAutoBucket, PlanNamedTypes::readAutoBucket),
             of(ScalarFunction.class, Case.class, PlanNamedTypes::writeCase, PlanNamedTypes::readCase),
@@ -863,7 +865,8 @@ public final class PlanNamedTypes {
         entry(name(ToIP.class), ToIP::new),
         entry(name(ToInteger.class), ToInteger::new),
         entry(name(ToLong.class), ToLong::new),
-        entry(name(ToString.class), ToString::new)
+        entry(name(ToString.class), ToString::new),
+        entry(name(ToVersion.class), ToVersion::new)
     );
 
     static UnaryScalarFunction readESQLUnaryScalar(PlanStreamInput in, String name) throws IOException {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/ComparisonMapper.java

@@ -110,8 +110,9 @@ abstract class ComparisonMapper<T extends BinaryComparison> extends EvalMapper.E
 
     @Override
     protected final Supplier<EvalOperator.ExpressionEvaluator> map(BinaryComparison bc, Layout layout) {
-        if (bc.left().dataType().isNumeric()) {
-            DataType type = EsqlDataTypeRegistry.INSTANCE.commonType(bc.left().dataType(), bc.right().dataType());
+        DataType leftType = bc.left().dataType();
+        if (leftType.isNumeric()) {
+            DataType type = EsqlDataTypeRegistry.INSTANCE.commonType(leftType, bc.right().dataType());
             if (type == DataTypes.INTEGER) {
                 return castToEvaluator(bc, layout, DataTypes.INTEGER, ints);
             }
@@ -124,13 +125,13 @@ abstract class ComparisonMapper<T extends BinaryComparison> extends EvalMapper.E
         }
         Supplier<EvalOperator.ExpressionEvaluator> leftEval = EvalMapper.toEvaluator(bc.left(), layout);
         Supplier<EvalOperator.ExpressionEvaluator> rightEval = EvalMapper.toEvaluator(bc.right(), layout);
-        if (bc.left().dataType() == DataTypes.KEYWORD || bc.left().dataType() == DataTypes.IP) {
+        if (leftType == DataTypes.KEYWORD || leftType == DataTypes.IP || leftType == DataTypes.VERSION) {
             return () -> keywords.apply(leftEval.get(), rightEval.get());
         }
-        if (bc.left().dataType() == DataTypes.BOOLEAN) {
+        if (leftType == DataTypes.BOOLEAN) {
             return () -> bools.apply(leftEval.get(), rightEval.get());
         }
-        if (bc.left().dataType() == DataTypes.DATETIME) {
+        if (leftType == DataTypes.DATETIME) {
             return () -> longs.apply(leftEval.get(), rightEval.get());
         }
         throw new AssertionError("resolved type for [" + bc + "] but didn't implement mapping");

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java

@@ -226,7 +226,10 @@ public class LocalExecutionPlanner {
             return ElementType.DOUBLE;
         }
         // unsupported fields are passed through as a BytesRef
-        if (dataType == DataTypes.KEYWORD || dataType == DataTypes.IP || dataType == DataTypes.UNSUPPORTED) {
+        if (dataType == DataTypes.KEYWORD
+            || dataType == DataTypes.IP
+            || dataType == DataTypes.VERSION
+            || dataType == DataTypes.UNSUPPORTED) {
             return ElementType.BYTES_REF;
         }
         if (dataType == DataTypes.NULL) {

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

@@ -35,6 +35,7 @@ import static org.elasticsearch.xpack.ql.type.DataTypes.OBJECT;
 import static org.elasticsearch.xpack.ql.type.DataTypes.SCALED_FLOAT;
 import static org.elasticsearch.xpack.ql.type.DataTypes.SHORT;
 import static org.elasticsearch.xpack.ql.type.DataTypes.UNSUPPORTED;
+import static org.elasticsearch.xpack.ql.type.DataTypes.VERSION;
 
 public final class EsqlDataTypes {
 
@@ -59,7 +60,8 @@ public final class EsqlDataTypes {
         IP,
         OBJECT,
         NESTED,
-        SCALED_FLOAT
+        SCALED_FLOAT,
+        VERSION
     ).sorted(Comparator.comparing(DataType::typeName)).toList();
 
     private static final Map<String, DataType> NAME_TO_TYPE = TYPES.stream().collect(toUnmodifiableMap(DataType::typeName, t -> t));

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/action/EsqlQueryResponseTests.java

@@ -28,6 +28,7 @@ import org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner;
 import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
 import org.elasticsearch.xpack.ql.type.DataType;
 import org.elasticsearch.xpack.ql.type.DataTypes;
+import org.elasticsearch.xpack.versionfield.Version;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -85,6 +86,7 @@ public class EsqlQueryResponseTests extends AbstractChunkedSerializingTestCase<E
                 case "unsupported" -> ((BytesRefBlock.Builder) builder).appendBytesRef(
                     new BytesRef(UnsupportedValueSource.UNSUPPORTED_OUTPUT)
                 );
+                case "version" -> ((BytesRefBlock.Builder) builder).appendBytesRef(new Version(randomIdentifier()).toBytesRef());
                 case "null" -> builder.appendNull();
                 default -> throw new UnsupportedOperationException("unsupported data type [" + c + "]");
             }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/AnalyzerTests.java

@@ -1166,15 +1166,14 @@ public class AnalyzerTests extends ESTestCase {
 
     public void testUnsupportedTypesWithToString() {
         // DATE_PERIOD and TIME_DURATION types have been added, but not really patched through the engine; i.e. supported.
+        final String supportedTypes = "boolean, datetime, double, integer, ip, keyword, long or version";
         verifyUnsupported(
             "row period = 1 year | eval to_string(period)",
-            "line 1:28: argument of [to_string(period)] must be [boolean, datetime, double, integer, ip, keyword or long], "
-                + "found value [period] type [date_period]"
+            "line 1:28: argument of [to_string(period)] must be [" + supportedTypes + "], found value [period] type [date_period]"
         );
         verifyUnsupported(
             "row duration = 1 hour | eval to_string(duration)",
-            "line 1:30: argument of [to_string(duration)] must be [boolean, datetime, double, integer, ip, keyword or long], "
-                + "found value [duration] type [time_duration]"
+            "line 1:30: argument of [to_string(duration)] must be [" + supportedTypes + "], found value [duration] type [time_duration]"
         );
         verifyUnsupported("from test | eval to_string(point)", "line 1:28: Cannot use field [point] with unsupported type [geo_point]");
     }

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

@@ -23,6 +23,7 @@ import org.elasticsearch.xpack.ql.expression.Literal;
 import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.ql.type.DataType;
 import org.elasticsearch.xpack.ql.type.EsField;
+import org.elasticsearch.xpack.versionfield.Version;
 import org.hamcrest.Matcher;
 
 import java.time.Duration;
@@ -62,6 +63,7 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
             case "keyword" -> new BytesRef(randomAlphaOfLength(5));
             case "ip" -> new BytesRef(InetAddressPoint.encode(randomIp(randomBoolean())));
             case "time_duration" -> Duration.ofMillis(randomNonNegativeLong());
+            case "version" -> new Version(randomIdentifier()).toBytesRef();
             case "null" -> null;
             default -> throw new IllegalArgumentException("can't make random values for [" + type.typeName() + "]");
         }, type);

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/predicate/operator/comparison/AbstractBinaryComparisonTestCase.java

@@ -70,7 +70,7 @@ public abstract class AbstractBinaryComparisonTestCase extends AbstractBinaryOpe
                 equalTo(
                     String.format(
                         Locale.ROOT,
-                        "first argument of [%s %s] must be [numeric, keyword, ip or datetime], found value [] type [%s]",
+                        "first argument of [%s %s] must be [numeric, keyword, ip, datetime or version], found value [] type [%s]",
                         lhsType.typeName(),
                         rhsType.typeName(),
                         lhsType.typeName()

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/planner/ExpressionTranslators.java

@@ -301,7 +301,9 @@ public final class ExpressionTranslators {
             } else if (field.dataType() == VERSION) {
                 // VersionStringFieldMapper#indexedValueForSearch() only accepts as input String or BytesRef with the String (i.e. not
                 // encoded) representation of the version as it'll do the encoding itself.
-                if (value instanceof Version version) {
+                if (value instanceof BytesRef bytesRef) {
+                    value = new Version(bytesRef).toString();
+                } else if (value instanceof Version version) {
                     value = version.toString();
                 }
             }