ESQL: TO_IP can handle leading zeros (#126532)

Modifies TO_IP so it can handle leading `0`s in ipv4s. Here's how it
works now:
```
ROW ip = TO_IP("192.168.0.1") // OK!
ROW ip = TO_IP("192.168.010.1") // Fails
```

This adds
```
ROW ip = TO_IP("192.168.010.1", {"leading_zeros": "octal"})
ROW ip = TO_IP("192.168.010.1", {"leading_zeros": "decimal"})
```

We do this because there isn't a consensus on how to parse leading zeros
in ipv4s. The standard unix tools like `ping` and `ftp` interpret
leading zeros as octal. Java's built in ip parsing interprets them as
decimal. Because folks are using this for security rules we need to
support all the choices.

Closes #125460

docs/changelog/126532.yaml

@@ -0,0 +1,6 @@
+pr: 126532
+summary: TO_IP can handle leading zeros
+area: ES|QL
+type: bug
+issues:
+ - 125460

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

@@ -1,6 +1,6 @@
 % This is generated by ESQL's AbstractFunctionTestCase. Do no edit it. See ../README.md for how to regenerate it.
 
-**Example**
+**Examples**
 
 ```esql
 ROW str1 = "1.1.1.1", str2 = "foo"
@@ -23,4 +23,26 @@ A following header will contain the failure reason and the offending value:
 
 `"java.lang.IllegalArgumentException: 'foo' is not an IP string literal."`
 
+```esql
+ROW s = "1.1.010.1" | EVAL ip = TO_IP(s, {"leading_zeros":"octal"})
+```
+
+| s:keyword | ip:ip |
+| --- | --- |
+| 1.1.010.1 | 1.1.8.1 |
+
+
+Parse v4 addresses with leading zeros as octal. Like `ping` or `ftp`.
+
+```esql
+ROW s = "1.1.010.1" | EVAL ip = TO_IP(s, {"leading_zeros":"decimal"})
+```
+
+| s:keyword | ip:ip |
+| --- | --- |
+| 1.1.010.1 | 1.1.10.1 |
+
+
+Parse v4 addresses with leading zeros as decimal. Java's `InetAddress.getByName`.
+
 

docs/reference/query-languages/esql/_snippets/functions/functionNamedParams/to_ip.md

@@ -0,0 +1,7 @@
+% This is generated by ESQL's AbstractFunctionTestCase. Do no edit it. See ../README.md for how to regenerate it.
+
+**Supported function named parameters**
+
+`leading_zeros`
+:   (keyword) What to do with leading 0s in IPv4 addresses.
+

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

@@ -19,5 +19,8 @@
 :::{include} ../types/to_ip.md
 :::
 
+:::{include} ../functionNamedParams/to_ip.md
+:::
+
 :::{include} ../examples/to_ip.md
 :::

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

@@ -5,3 +5,6 @@
 `field`
 :   Input value. The input can be a single- or multi-valued column or an expression.
 
+`options`
+:   (Optional) Additional options.
+

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

@@ -2,9 +2,9 @@
 
 **Supported types**
 
-| field | result |
-| --- | --- |
-| ip | ip |
-| keyword | ip |
-| text | ip |
+| field | options | result |
+| --- | --- | --- |
+| ip | | ip |
+| keyword | | ip |
+| text | | ip |
 

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

@@ -1 +1 @@
-<svg version="1.1" xmlns:xlink="http://www.w3.org/1999/xlink" xmlns="http://www.w3.org/2000/svg" width="264" height="46" viewbox="0 0 264 46"><defs><style type="text/css">.c{fill:none;stroke:#222222;}.k{fill:#000000;font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;font-size:20px;}.s{fill:#e4f4ff;stroke:#222222;}.syn{fill:#8D8D8D;font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;font-size:20px;}</style></defs><path class="c" d="M0 31h5m80 0h10m32 0h10m80 0h10m32 0h5"/><rect class="s" x="5" y="5" width="80" height="36"/><text class="k" x="15" y="31">TO_IP</text><rect class="s" x="95" y="5" width="32" height="36" rx="7"/><text class="syn" x="105" y="31">(</text><rect class="s" x="137" y="5" width="80" height="36" rx="7"/><text class="k" x="147" y="31">field</text><rect class="s" x="227" y="5" width="32" height="36" rx="7"/><text class="syn" x="237" y="31">)</text></svg>
+<svg version="1.1" xmlns:xlink="http://www.w3.org/1999/xlink" xmlns="http://www.w3.org/2000/svg" width="460" height="61" viewbox="0 0 460 61"><defs><style type="text/css">.c{fill:none;stroke:#222222;}.k{fill:#000000;font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;font-size:20px;}.s{fill:#e4f4ff;stroke:#222222;}.syn{fill:#8D8D8D;font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;font-size:20px;}</style></defs><path class="c" d="M0 31h5m80 0h10m32 0h10m80 0h10m32 0h30m104 0h20m-139 0q5 0 5 5v10q0 5 5 5h114q5 0 5-5v-10q0-5 5-5m5 0h10m32 0h5"/><rect class="s" x="5" y="5" width="80" height="36"/><text class="k" x="15" y="31">TO_IP</text><rect class="s" x="95" y="5" width="32" height="36" rx="7"/><text class="syn" x="105" y="31">(</text><rect class="s" x="137" y="5" width="80" height="36" rx="7"/><text class="k" x="147" y="31">field</text><rect class="s" x="227" y="5" width="32" height="36" rx="7"/><text class="syn" x="237" y="31">,</text><rect class="s" x="289" y="5" width="104" height="36" rx="7"/><text class="k" x="299" y="31">options</text><rect class="s" x="423" y="5" width="32" height="36" rx="7"/><text class="syn" x="433" y="31">)</text></svg>

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

@@ -42,7 +42,9 @@
     }
   ],
   "examples" : [
-    "ROW str1 = \"1.1.1.1\", str2 = \"foo\"\n| EVAL ip1 = TO_IP(str1), ip2 = TO_IP(str2)\n| WHERE CIDR_MATCH(ip1, \"1.0.0.0/8\")"
+    "ROW str1 = \"1.1.1.1\", str2 = \"foo\"\n| EVAL ip1 = TO_IP(str1), ip2 = TO_IP(str2)\n| WHERE CIDR_MATCH(ip1, \"1.0.0.0/8\")",
+    "ROW s = \"1.1.010.1\" | EVAL ip = TO_IP(s, {\"leading_zeros\":\"octal\"})",
+    "ROW s = \"1.1.010.1\" | EVAL ip = TO_IP(s, {\"leading_zeros\":\"decimal\"})"
   ],
   "preview" : false,
   "snapshot_only" : false

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

@@ -133,6 +133,7 @@ public class CsvTestsDataLoader {
     private static final TestDataset ADDRESSES = new TestDataset("addresses");
     private static final TestDataset BOOKS = new TestDataset("books").withSetting("books-settings.json");
     private static final TestDataset SEMANTIC_TEXT = new TestDataset("semantic_text").withInferenceEndpoint(true);
+    private static final TestDataset LOGS = new TestDataset("logs");
 
     public static final Map<String, TestDataset> CSV_DATASET_MAP = Map.ofEntries(
         Map.entry(EMPLOYEES.indexName, EMPLOYEES),
@@ -182,7 +183,8 @@ public class CsvTestsDataLoader {
         Map.entry(DISTANCES.indexName, DISTANCES),
         Map.entry(ADDRESSES.indexName, ADDRESSES),
         Map.entry(BOOKS.indexName, BOOKS),
-        Map.entry(SEMANTIC_TEXT.indexName, SEMANTIC_TEXT)
+        Map.entry(SEMANTIC_TEXT.indexName, SEMANTIC_TEXT),
+        Map.entry(LOGS.indexName, LOGS)
     );
 
     private static final EnrichConfig LANGUAGES_ENRICH = new EnrichConfig("languages_policy", "enrich-policy-languages.json");

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

@@ -0,0 +1,5 @@
+@timestamp:date         ,system:keyword,message:keyword
+2023-10-23T13:55:01.543Z,          ping,Pinging 192.168.86.046
+2023-10-23T13:55:01.544Z,          cron,Running cats
+2023-10-23T13:55:01.545Z,          java,Doing java stuff for 192.168.86.038
+2023-10-23T13:55:01.546Z,          java,More java stuff

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

@@ -271,6 +271,86 @@ str1:keyword |str2:keyword |ip1:ip  |ip2:ip
 // end::to_ip-result[]
 ;
 
+convertFromStringLeadingZerosDecimal
+required_capability: to_ip_leading_zeros
+// tag::to_ip_leading_zeros_decimal[]
+ROW s = "1.1.010.1" | EVAL ip = TO_IP(s, {"leading_zeros":"decimal"})
+// end::to_ip_leading_zeros_decimal[]
+;
+
+// tag::to_ip_leading_zeros_decimal-result[]
+s:keyword | ip:ip
+1.1.010.1 | 1.1.10.1
+// end::to_ip_leading_zeros_decimal-result[]
+;
+
+convertFromStringLeadingZerosOctal
+required_capability: to_ip_leading_zeros
+// tag::to_ip_leading_zeros_octal[]
+ROW s = "1.1.010.1" | EVAL ip = TO_IP(s, {"leading_zeros":"octal"})
+// end::to_ip_leading_zeros_octal[]
+;
+
+// tag::to_ip_leading_zeros_octal-result[]
+s:keyword | ip:ip
+1.1.010.1 | 1.1.8.1
+// end::to_ip_leading_zeros_octal-result[]
+;
+
+convertFromStringFancy
+required_capability: to_ip_leading_zeros
+FROM logs
+| KEEP @timestamp, system, message
+| EVAL client = CASE(
+    system == "ping",
+        TO_IP(REPLACE(message, "Pinging ", ""), {"leading_zeros": "octal"}),
+    system == "java" AND STARTS_WITH(message, "Doing java stuff for "),
+        TO_IP(REPLACE(message, "Doing java stuff for ", ""), {"leading_zeros": "decimal"}))
+| SORT @timestamp
+| LIMIT 4
+;
+
+@timestamp:date         |system:keyword|message:keyword                    |client:ip
+2023-10-23T13:55:01.543Z|          ping|Pinging 192.168.86.046             |192.168.86.38
+2023-10-23T13:55:01.544Z|          cron|Running cats                       |null
+2023-10-23T13:55:01.545Z|          java|Doing java stuff for 192.168.86.038|192.168.86.38
+2023-10-23T13:55:01.546Z|          java|More java stuff                    |null
+;
+
+toIpInAgg
+ROW s = "1.1.1.1" | STATS COUNT(*) BY ip = TO_IP(s)
+;
+
+COUNT(*):long | ip:ip
+            1 | 1.1.1.1
+;
+
+toIpInSort
+ROW s = "1.1.1.1" | SORT TO_IP(s)
+;
+
+s:keyword
+1.1.1.1
+;
+
+toIpInAggOctal
+required_capability: to_ip_leading_zeros
+ROW s = "1.1.010.1" | STATS COUNT(*) BY ip = TO_IP(s, {"leading_zeros":"octal"})
+;
+
+COUNT(*):long | ip:ip
+            1 | 1.1.8.1
+;
+
+toIpInSortOctal
+required_capability: to_ip_leading_zeros
+ROW s = "1.1.010.1" | SORT TO_IP(s, {"leading_zeros":"octal"})
+;
+
+s:keyword
+1.1.010.1
+;
+
 cdirMatchOrsIPs
 required_capability: combine_disjunctive_cidrmatches
 

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

@@ -0,0 +1,13 @@
+{
+    "properties" : {
+        "@timestamp" : {
+            "type" : "date"
+        },
+        "system" : {
+          "type" : "keyword"
+        },
+        "message" : {
+            "type" : "keyword"
+        }
+    }
+}

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

@@ -982,6 +982,11 @@ public class EsqlCapabilities {
          */
         FORK_V2(Build.current().isSnapshot()),
 
+        /**
+         * Support for the {@code leading_zeros} named parameter.
+         */
+        TO_IP_LEADING_ZEROS,
+
         /**
          * Does the usage information for ESQL contain a histogram of {@code took} values?
          */

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

@@ -58,6 +58,7 @@ import org.elasticsearch.xpack.esql.expression.function.scalar.conditional.Case;
 import org.elasticsearch.xpack.esql.expression.function.scalar.conditional.Greatest;
 import org.elasticsearch.xpack.esql.expression.function.scalar.conditional.Least;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.AbstractConvertFunction;
+import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ConvertFunction;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.FoldablesConvertFunction;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToDouble;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToInteger;
@@ -70,6 +71,7 @@ import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.In;
 import org.elasticsearch.xpack.esql.index.EsIndex;
 import org.elasticsearch.xpack.esql.index.IndexResolution;
 import org.elasticsearch.xpack.esql.inference.ResolvedInference;
+import org.elasticsearch.xpack.esql.optimizer.rules.logical.SubstituteSurrogateExpressions;
 import org.elasticsearch.xpack.esql.parser.ParsingException;
 import org.elasticsearch.xpack.esql.plan.IndexPattern;
 import org.elasticsearch.xpack.esql.plan.logical.Aggregate;
@@ -1578,10 +1580,12 @@ public class Analyzer extends ParameterizedRuleExecutor<LogicalPlan, AnalyzerCon
             int alreadyAddedUnionFieldAttributes = unionFieldAttributes.size();
             // See if the eval function has an unresolved MultiTypeEsField field
             // Replace the entire convert function with a new FieldAttribute (containing type conversion knowledge)
-            plan = plan.transformExpressionsOnly(
-                AbstractConvertFunction.class,
-                convert -> resolveConvertFunction(convert, unionFieldAttributes)
-            );
+            plan = plan.transformExpressionsOnly(e -> {
+                if (e instanceof ConvertFunction convert) {
+                    return resolveConvertFunction(convert, unionFieldAttributes);
+                }
+                return e;
+            });
             // If no union fields were generated, return the plan as is
             if (unionFieldAttributes.size() == alreadyAddedUnionFieldAttributes) {
                 return plan;
@@ -1612,7 +1616,8 @@ public class Analyzer extends ParameterizedRuleExecutor<LogicalPlan, AnalyzerCon
             return plan;
         }
 
-        private Expression resolveConvertFunction(AbstractConvertFunction convert, List<FieldAttribute> unionFieldAttributes) {
+        private Expression resolveConvertFunction(ConvertFunction convert, List<FieldAttribute> unionFieldAttributes) {
+            Expression convertExpression = (Expression) convert;
             if (convert.field() instanceof FieldAttribute fa && fa.field() instanceof InvalidMappedField imf) {
                 HashMap<TypeResolutionKey, Expression> typeResolutions = new HashMap<>();
                 Set<DataType> supportedTypes = convert.supportedTypes();
@@ -1639,9 +1644,11 @@ public class Analyzer extends ParameterizedRuleExecutor<LogicalPlan, AnalyzerCon
                     return createIfDoesNotAlreadyExist(fa, resolvedField, unionFieldAttributes);
                 }
             } else if (convert.field() instanceof AbstractConvertFunction subConvert) {
-                return convert.replaceChildren(Collections.singletonList(resolveConvertFunction(subConvert, unionFieldAttributes)));
+                return convertExpression.replaceChildren(
+                    Collections.singletonList(resolveConvertFunction(subConvert, unionFieldAttributes))
+                );
             }
-            return convert;
+            return convertExpression;
         }
 
         private Expression createIfDoesNotAlreadyExist(
@@ -1677,7 +1684,7 @@ public class Analyzer extends ParameterizedRuleExecutor<LogicalPlan, AnalyzerCon
             return MultiTypeEsField.resolveFrom(imf, typesToConversionExpressions);
         }
 
-        private Expression typeSpecificConvert(AbstractConvertFunction convert, Source source, DataType type, InvalidMappedField mtf) {
+        private Expression typeSpecificConvert(ConvertFunction convert, Source source, DataType type, InvalidMappedField mtf) {
             EsField field = new EsField(mtf.getName(), type, mtf.getProperties(), mtf.isAggregatable());
             FieldAttribute originalFieldAttr = (FieldAttribute) convert.field();
             FieldAttribute resolvedAttr = new FieldAttribute(
@@ -1689,7 +1696,13 @@ public class Analyzer extends ParameterizedRuleExecutor<LogicalPlan, AnalyzerCon
                 originalFieldAttr.id(),
                 true
             );
-            return convert.replaceChildren(Collections.singletonList(resolvedAttr));
+            Expression e = ((Expression) convert).replaceChildren(Collections.singletonList(resolvedAttr));
+            /*
+             * Resolve surrogates immediately because these type specific conversions are serialized
+             * and SurrogateExpressions are expected to be resolved on the coordinating node. At least,
+             * TO_IP is expected to be resolved there.
+             */
+            return SubstituteSurrogateExpressions.rule(e);
         }
     }
 

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

@@ -25,8 +25,10 @@ import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToDegrees
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToDouble;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToGeoPoint;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToGeoShape;
-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.ToIpLeadingZerosDecimal;
+import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToIpLeadingZerosOctal;
+import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToIpLeadingZerosRejected;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToLong;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToRadians;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToString;
@@ -192,7 +194,9 @@ public class ExpressionWritables {
         entries.add(ToGeoShape.ENTRY);
         entries.add(ToCartesianShape.ENTRY);
         entries.add(ToGeoPoint.ENTRY);
-        entries.add(ToIP.ENTRY);
+        entries.add(ToIpLeadingZerosDecimal.ENTRY);
+        entries.add(ToIpLeadingZerosOctal.ENTRY);
+        entries.add(ToIpLeadingZerosRejected.ENTRY);
         entries.add(ToInteger.ENTRY);
         entries.add(ToLong.ENTRY);
         entries.add(ToRadians.ENTRY);

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

@@ -16,6 +16,7 @@ import org.elasticsearch.xpack.esql.core.expression.function.Function;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.util.Check;
+import org.elasticsearch.xpack.esql.expression.SurrogateExpression;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Avg;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.AvgOverTime;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Count;
@@ -56,8 +57,11 @@ import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToDegrees
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToDouble;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToGeoPoint;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToGeoShape;
-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.ToIp;
+import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToIpLeadingZerosDecimal;
+import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToIpLeadingZerosOctal;
+import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToIpLeadingZerosRejected;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToLong;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToRadians;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToString;
@@ -230,6 +234,7 @@ public class EsqlFunctionRegistry {
     public EsqlFunctionRegistry() {
         register(functions());
         buildDataTypesForStringLiteralConversion(functions());
+        nameSurrogates();
     }
 
     EsqlFunctionRegistry(FunctionDefinition... functions) {
@@ -391,7 +396,7 @@ public class EsqlFunctionRegistry {
                 def(ToDouble.class, ToDouble::new, "to_double", "to_dbl"),
                 def(ToGeoPoint.class, ToGeoPoint::new, "to_geopoint"),
                 def(ToGeoShape.class, ToGeoShape::new, "to_geoshape"),
-                def(ToIP.class, ToIP::new, "to_ip"),
+                def(ToIp.class, ToIp::new, "to_ip"),
                 def(ToInteger.class, ToInteger::new, "to_integer", "to_int"),
                 def(ToLong.class, ToLong::new, "to_long"),
                 def(ToRadians.class, ToRadians::new, "to_radians"),
@@ -795,6 +800,15 @@ public class EsqlFunctionRegistry {
         }
     }
 
+    /**
+     * Add {@link #names} entries for functions that are not registered, but we rewrite to using {@link SurrogateExpression}.
+     */
+    private void nameSurrogates() {
+        names.put(ToIpLeadingZerosRejected.class, "TO_IP");
+        names.put(ToIpLeadingZerosDecimal.class, "TO_IP");
+        names.put(ToIpLeadingZerosOctal.class, "TO_IP");
+    }
+
     protected interface FunctionBuilder {
         Function build(Source source, List<Expression> children, Configuration cfg);
     }

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

@@ -43,7 +43,7 @@ import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isTyp
  *     {@link org.elasticsearch.xpack.esql.expression.function.scalar}.
  * </p>
  */
-public abstract class AbstractConvertFunction extends UnaryScalarFunction {
+public abstract class AbstractConvertFunction extends UnaryScalarFunction implements ConvertFunction {
 
     // the numeric types convert functions need to handle; the other numeric types are converted upstream to one of these
     private static final List<DataType> NUMERIC_TYPES = List.of(DataType.INTEGER, DataType.LONG, DataType.UNSIGNED_LONG, DataType.DOUBLE);
@@ -76,11 +76,12 @@ public abstract class AbstractConvertFunction extends UnaryScalarFunction {
         return isTypeOrUnionType(field(), factories()::containsKey, sourceText(), null, supportedTypesNames(supportedTypes()));
     }
 
+    @Override
     public Set<DataType> supportedTypes() {
         return factories().keySet();
     }
 
-    private static String supportedTypesNames(Set<DataType> types) {
+    static String supportedTypesNames(Set<DataType> types) {
         List<String> supportedTypesNames = new ArrayList<>(types.size());
         HashSet<DataType> supportTypes = new HashSet<>(types);
         if (supportTypes.containsAll(NUMERIC_TYPES)) {

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

@@ -0,0 +1,28 @@
+/*
+ * 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.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+
+import java.util.Set;
+
+/**
+ * A function that converts from one type to another.
+ */
+public interface ConvertFunction {
+    /**
+     * Expression containing the values to be converted.
+     */
+    Expression field();
+
+    /**
+     * The types that {@link #field()} can have.
+     */
+    Set<DataType> supportedTypes();
+}

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

@@ -42,6 +42,14 @@ public class ParseIp {
         return new ParseIpLeadingZerosRejectedEvaluator.Factory(source, field, driverContext -> buildScratch(driverContext.breaker()));
     };
 
+    static final AbstractConvertFunction.BuildFactory FROM_KEYWORD_LEADING_ZEROS_DECIMAL = (source, field) -> {
+        return new ParseIpLeadingZerosAreDecimalEvaluator.Factory(source, field, driverContext -> buildScratch(driverContext.breaker()));
+    };
+
+    static final AbstractConvertFunction.BuildFactory FROM_KEYWORD_LEADING_ZEROS_OCTAL = (source, field) -> {
+        return new ParseIpLeadingZerosAreOctalEvaluator.Factory(source, field, driverContext -> buildScratch(driverContext.breaker()));
+    };
+
     public static BreakingBytesRefBuilder buildScratch(CircuitBreaker breaker) {
         BreakingBytesRefBuilder scratch = new BreakingBytesRefBuilder(breaker, "to_ip", 16);
         scratch.setLength(InetAddressPoint.BYTES);

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

@@ -0,0 +1,255 @@
+/*
+ * 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.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.compute.operator.EvalOperator;
+import org.elasticsearch.xpack.esql.core.expression.EntryExpression;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.expression.Literal;
+import org.elasticsearch.xpack.esql.core.expression.MapExpression;
+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.SurrogateExpression;
+import org.elasticsearch.xpack.esql.expression.function.Example;
+import org.elasticsearch.xpack.esql.expression.function.FunctionInfo;
+import org.elasticsearch.xpack.esql.expression.function.MapParam;
+import org.elasticsearch.xpack.esql.expression.function.OptionalArgument;
+import org.elasticsearch.xpack.esql.expression.function.Param;
+import org.elasticsearch.xpack.esql.expression.function.scalar.EsqlScalarFunction;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.ParamOrdinal.SECOND;
+import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isMapExpression;
+import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isTypeOrUnionType;
+import static org.elasticsearch.xpack.esql.core.type.DataType.IP;
+import static org.elasticsearch.xpack.esql.core.type.DataType.KEYWORD;
+import static org.elasticsearch.xpack.esql.expression.function.scalar.convert.AbstractConvertFunction.supportedTypesNames;
+
+/**
+ * Converts strings to IPs.
+ * <p>
+ *     IPv4 addresses have traditionally parsed quads with leading zeros in three
+ *     mutually exclusive ways:
+ * </p>
+ * <ul>
+ *     <li>As octal numbers. So {@code 1.1.010.1} becomes {@code 1.1.8.1}.</li>
+ *     <li>As decimal numbers. So {@code 1.1.010.1} becomes {@code 1.1.10.1}.</li>
+ *     <li>Rejects them entirely. So {@code 1.1.010.1} becomes {@code null} with a warning.</li>
+ * </ul>
+ * <p>
+ *     These three ways of handling leading zeros are available with the optional
+ *     {@code leading_zeros} named parameter. Set to {@code octal}, {@code decimal},
+ *     or {@code reject}. If not sent this defaults to {@code reject} which has
+ *     been Elasticsearch's traditional way of handling leading zeros for years.
+ * </p>
+ * <p>
+ *     This doesn't extend from {@link AbstractConvertFunction} so that it can
+ *     support a named parameter for the leading zeros behavior. Instead, it rewrites
+ *     itself into either {@link ToIpLeadingZerosOctal}, {@link ToIpLeadingZerosDecimal},
+ *     or {@link ToIpLeadingZerosRejected} which are all {@link AbstractConvertFunction}
+ *     subclasses. This keeps the conversion code happy while still allowing us to
+ *     expose a single method to users.
+ * </p>
+ */
+public class ToIp extends EsqlScalarFunction implements SurrogateExpression, OptionalArgument, ConvertFunction {
+    private static final String LEADING_ZEROS = "leading_zeros";
+    public static final Map<String, DataType> ALLOWED_OPTIONS = Map.ofEntries(Map.entry(LEADING_ZEROS, KEYWORD));
+
+    private final Expression field;
+    private final Expression options;
+
+    @FunctionInfo(
+        returnType = "ip",
+        description = "Converts an input string to an IP value.",
+        examples = {
+            @Example(file = "ip", tag = "to_ip", explanation = """
+                Note that in this example, the last conversion of the string isn’t possible.
+                When this happens, the result is a `null` value. In this case a _Warning_ header is added to the response.
+                The header will provide information on the source of the failure:
+
+                `"Line 1:68: evaluation of [TO_IP(str2)] failed, treating result as null. Only first 20 failures recorded."`
+
+                A following header will contain the failure reason and the offending value:
+
+                `"java.lang.IllegalArgumentException: 'foo' is not an IP string literal."`"""),
+            @Example(file = "ip", tag = "to_ip_leading_zeros_octal", explanation = """
+                Parse v4 addresses with leading zeros as octal. Like `ping` or `ftp`.
+                """),
+            @Example(file = "ip", tag = "to_ip_leading_zeros_decimal", explanation = """
+                Parse v4 addresses with leading zeros as decimal. Java's `InetAddress.getByName`.
+                """) }
+    )
+    public ToIp(
+        Source source,
+        @Param(
+            name = "field",
+            type = { "ip", "keyword", "text" },
+            description = "Input value. The input can be a single- or multi-valued column or an expression."
+        ) Expression field,
+        @MapParam(
+            name = "options",
+            params = {
+                @MapParam.MapParamEntry(
+                    name = "leading_zeros",
+                    type = "keyword",
+                    valueHint = { "reject", "octal", "decimal" },
+                    description = "What to do with leading 0s in IPv4 addresses."
+                ) },
+            description = "(Optional) Additional options.",
+            optional = true
+        ) Expression options
+    ) {
+        super(source, options == null ? List.of(field) : List.of(field, options));
+        this.field = field;
+        this.options = options;
+    }
+
+    @Override
+    public String getWriteableName() {
+        throw new UnsupportedOperationException("not serialized");
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        throw new UnsupportedOperationException("not serialized");
+    }
+
+    @Override
+    public DataType dataType() {
+        return IP;
+    }
+
+    @Override
+    public Expression replaceChildren(List<Expression> newChildren) {
+        return new ToIp(source(), newChildren.get(0), newChildren.size() == 1 ? null : newChildren.get(1));
+    }
+
+    @Override
+    protected NodeInfo<? extends Expression> info() {
+        return NodeInfo.create(this, ToIp::new, field, options);
+    }
+
+    @Override
+    public EvalOperator.ExpressionEvaluator.Factory toEvaluator(ToEvaluator toEvaluator) {
+        throw new UnsupportedOperationException("should be rewritten");
+    }
+
+    @Override
+    public Expression surrogate() {
+        return LeadingZeros.from((MapExpression) options).surrogate(source(), field);
+    }
+
+    @Override
+    public Expression field() {
+        return field;
+    }
+
+    @Override
+    public Set<DataType> supportedTypes() {
+        // All ToIpLeadingZeros* functions support the same input set. So we just pick one.
+        return ToIpLeadingZerosRejected.EVALUATORS.keySet();
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        if (childrenResolved() == false) {
+            return new TypeResolution("Unresolved children");
+        }
+        TypeResolution resolution = isTypeOrUnionType(
+            field,
+            ToIpLeadingZerosRejected.EVALUATORS::containsKey,
+            sourceText(),
+            null,
+            supportedTypesNames(supportedTypes())
+        );
+        if (resolution.unresolved()) {
+            return resolution;
+        }
+        if (options == null) {
+            return resolution;
+        }
+        resolution = isMapExpression(options, sourceText(), SECOND);
+        if (resolution.unresolved()) {
+            return resolution;
+        }
+        for (EntryExpression e : ((MapExpression) options).entryExpressions()) {
+            String key;
+            if (e.key().dataType() != KEYWORD) {
+                return new TypeResolution("map keys must be strings");
+            }
+            if (e.key() instanceof Literal keyl) {
+                key = (String) keyl.value();
+            } else {
+                return new TypeResolution("map keys must be literals");
+            }
+            DataType expected = ALLOWED_OPTIONS.get(key);
+            if (expected == null) {
+                return new TypeResolution("[" + key + "] is not a supported option");
+            }
+
+            if (e.value().dataType() != expected) {
+                return new TypeResolution("[" + key + "] expects [" + expected + "] but was [" + e.value().dataType() + "]");
+            }
+            if (e.value() instanceof Literal == false) {
+                return new TypeResolution("map values must be literals");
+            }
+        }
+        try {
+            LeadingZeros.from((MapExpression) options);
+        } catch (IllegalArgumentException e) {
+            return new TypeResolution(e.getMessage());
+        }
+        return TypeResolution.TYPE_RESOLVED;
+    }
+
+    public enum LeadingZeros {
+        REJECT {
+            @Override
+            public Expression surrogate(Source source, Expression field) {
+                return new ToIpLeadingZerosRejected(source, field);
+            }
+        },
+        DECIMAL {
+            @Override
+            public Expression surrogate(Source source, Expression field) {
+                return new ToIpLeadingZerosDecimal(source, field);
+            }
+        },
+        OCTAL {
+            @Override
+            public Expression surrogate(Source source, Expression field) {
+                return new ToIpLeadingZerosOctal(source, field);
+            }
+        };
+
+        public static LeadingZeros from(MapExpression exp) {
+            if (exp == null) {
+                return REJECT;
+            }
+            Expression e = exp.keyFoldedMap().get(LEADING_ZEROS);
+            return e == null ? REJECT : from((String) ((Literal) e).value());
+        }
+
+        public static LeadingZeros from(String str) {
+            return switch (str) {
+                case "reject" -> REJECT;
+                case "octal" -> OCTAL;
+                case "decimal" -> DECIMAL;
+                default -> throw new IllegalArgumentException("Illegal leading_zeros [" + str + "]");
+            };
+        }
+
+        public abstract Expression surrogate(Source source, Expression field);
+    }
+}

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

@@ -0,0 +1,71 @@
+/*
+ * 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.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.tree.NodeInfo;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.xpack.esql.core.type.DataType.IP;
+import static org.elasticsearch.xpack.esql.core.type.DataType.KEYWORD;
+import static org.elasticsearch.xpack.esql.core.type.DataType.TEXT;
+import static org.elasticsearch.xpack.esql.expression.function.scalar.convert.ParseIp.FROM_KEYWORD_LEADING_ZEROS_DECIMAL;
+
+public class ToIpLeadingZerosDecimal extends AbstractConvertFunction {
+    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
+        Expression.class,
+        "ToIpLeadingZerosDecimal",
+        ToIpLeadingZerosDecimal::new
+    );
+
+    static final Map<DataType, BuildFactory> EVALUATORS = Map.ofEntries(
+        Map.entry(IP, (source, field) -> field),
+        Map.entry(KEYWORD, FROM_KEYWORD_LEADING_ZEROS_DECIMAL),
+        Map.entry(TEXT, FROM_KEYWORD_LEADING_ZEROS_DECIMAL)
+    );
+
+    public ToIpLeadingZerosDecimal(Source source, Expression field) {
+        super(source, field);
+    }
+
+    private ToIpLeadingZerosDecimal(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return ENTRY.name;
+    }
+
+    @Override
+    protected Map<DataType, BuildFactory> factories() {
+        return EVALUATORS;
+    }
+
+    @Override
+    public DataType dataType() {
+        return IP;
+    }
+
+    @Override
+    public Expression replaceChildren(List<Expression> newChildren) {
+        return new ToIpLeadingZerosDecimal(source(), newChildren.get(0));
+    }
+
+    @Override
+    protected NodeInfo<? extends Expression> info() {
+        return NodeInfo.create(this, ToIpLeadingZerosDecimal::new, field());
+    }
+}

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

@@ -0,0 +1,71 @@
+/*
+ * 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.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.tree.NodeInfo;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.xpack.esql.core.type.DataType.IP;
+import static org.elasticsearch.xpack.esql.core.type.DataType.KEYWORD;
+import static org.elasticsearch.xpack.esql.core.type.DataType.TEXT;
+import static org.elasticsearch.xpack.esql.expression.function.scalar.convert.ParseIp.FROM_KEYWORD_LEADING_ZEROS_OCTAL;
+
+public class ToIpLeadingZerosOctal extends AbstractConvertFunction {
+    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
+        Expression.class,
+        "ToIpLeadingZerosOctal",
+        ToIpLeadingZerosOctal::new
+    );
+
+    static final Map<DataType, BuildFactory> EVALUATORS = Map.ofEntries(
+        Map.entry(IP, (source, field) -> field),
+        Map.entry(KEYWORD, FROM_KEYWORD_LEADING_ZEROS_OCTAL),
+        Map.entry(TEXT, FROM_KEYWORD_LEADING_ZEROS_OCTAL)
+    );
+
+    public ToIpLeadingZerosOctal(Source source, Expression field) {
+        super(source, field);
+    }
+
+    private ToIpLeadingZerosOctal(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return ENTRY.name;
+    }
+
+    @Override
+    protected Map<DataType, BuildFactory> factories() {
+        return EVALUATORS;
+    }
+
+    @Override
+    public DataType dataType() {
+        return IP;
+    }
+
+    @Override
+    public Expression replaceChildren(List<Expression> newChildren) {
+        return new ToIpLeadingZerosOctal(source(), newChildren.get(0));
+    }
+
+    @Override
+    protected NodeInfo<? extends Expression> info() {
+        return NodeInfo.create(this, ToIpLeadingZerosOctal::new, field());
+    }
+}

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

@@ -13,9 +13,6 @@ import org.elasticsearch.xpack.esql.core.expression.Expression;
 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.FunctionInfo;
-import org.elasticsearch.xpack.esql.expression.function.Param;
 
 import java.io.IOException;
 import java.util.List;
@@ -26,41 +23,28 @@ import static org.elasticsearch.xpack.esql.core.type.DataType.KEYWORD;
 import static org.elasticsearch.xpack.esql.core.type.DataType.TEXT;
 import static org.elasticsearch.xpack.esql.expression.function.scalar.convert.ParseIp.FROM_KEYWORD_LEADING_ZEROS_REJECTED;
 
-public class ToIP extends AbstractConvertFunction {
-    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(Expression.class, "ToIP", ToIP::new);
+public class ToIpLeadingZerosRejected extends AbstractConvertFunction {
+    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
+        Expression.class,
+        /*
+         * This is the name a function with this behavior has had since the
+         * dawn of ESQL. The ToIp function that exists now is not serialized.
+         */
+        "ToIP",
+        ToIpLeadingZerosRejected::new
+    );
 
-    private static final Map<DataType, BuildFactory> EVALUATORS = Map.ofEntries(
+    static final Map<DataType, BuildFactory> EVALUATORS = Map.ofEntries(
         Map.entry(IP, (source, field) -> field),
         Map.entry(KEYWORD, FROM_KEYWORD_LEADING_ZEROS_REJECTED),
         Map.entry(TEXT, FROM_KEYWORD_LEADING_ZEROS_REJECTED)
     );
 
-    @FunctionInfo(
-        returnType = "ip",
-        description = "Converts an input string to an IP value.",
-        examples = @Example(file = "ip", tag = "to_ip", explanation = """
-            Note that in this example, the last conversion of the string isn’t possible.
-            When this happens, the result is a `null` value. In this case a _Warning_ header is added to the response.
-            The header will provide information on the source of the failure:
-
-            `"Line 1:68: evaluation of [TO_IP(str2)] failed, treating result as null. Only first 20 failures recorded."`
-
-            A following header will contain the failure reason and the offending value:
-
-            `"java.lang.IllegalArgumentException: 'foo' is not an IP string literal."`""")
-    )
-    public ToIP(
-        Source source,
-        @Param(
-            name = "field",
-            type = { "ip", "keyword", "text" },
-            description = "Input value. The input can be a single- or multi-valued column or an expression."
-        ) Expression field
-    ) {
+    public ToIpLeadingZerosRejected(Source source, Expression field) {
         super(source, field);
     }
 
-    private ToIP(StreamInput in) throws IOException {
+    private ToIpLeadingZerosRejected(StreamInput in) throws IOException {
         super(in);
     }
 
@@ -81,11 +65,11 @@ public class ToIP extends AbstractConvertFunction {
 
     @Override
     public Expression replaceChildren(List<Expression> newChildren) {
-        return new ToIP(source(), newChildren.get(0));
+        return new ToIpLeadingZerosRejected(source(), newChildren.get(0));
     }
 
     @Override
     protected NodeInfo<? extends Expression> info() {
-        return NodeInfo.create(this, ToIP::new, field());
+        return NodeInfo.create(this, ToIpLeadingZerosRejected::new, field());
     }
 }

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

@@ -34,6 +34,7 @@ import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.util.Check;
 import org.elasticsearch.xpack.esql.core.util.SpatialCoordinateTypes;
 import org.elasticsearch.xpack.esql.evaluator.mapper.EvaluatorMapper;
+import org.elasticsearch.xpack.esql.expression.SurrogateExpression;
 import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.LucenePushdownPredicates;
 import org.elasticsearch.xpack.esql.planner.TranslatorHandler;
 import org.elasticsearch.xpack.esql.querydsl.query.SpatialRelatesQuery;
@@ -48,7 +49,8 @@ public abstract class SpatialRelatesFunction extends BinarySpatialFunction
     implements
         EvaluatorMapper,
         SpatialEvaluatorFactory.SpatialSourceSupplier,
-        TranslationAware {
+        TranslationAware,
+        SurrogateExpression {
 
     protected SpatialRelatesFunction(Source source, Expression left, Expression right, boolean leftDocValues, boolean rightDocValues) {
         super(source, left, right, leftDocValues, rightDocValues, false);
@@ -73,6 +75,7 @@ public abstract class SpatialRelatesFunction extends BinarySpatialFunction
     /**
      * Some spatial functions can replace themselves with alternatives that are more efficient for certain cases.
      */
+    @Override
     public SpatialRelatesFunction surrogate() {
         return this;
     }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizer.java

@@ -58,9 +58,9 @@ import org.elasticsearch.xpack.esql.optimizer.rules.logical.SkipQueryOnEmptyMapp
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.SkipQueryOnLimitZero;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.SplitInWithFoldableValue;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.SubstituteFilteredExpression;
-import org.elasticsearch.xpack.esql.optimizer.rules.logical.SubstituteSpatialSurrogates;
+import org.elasticsearch.xpack.esql.optimizer.rules.logical.SubstituteSurrogateAggregations;
+import org.elasticsearch.xpack.esql.optimizer.rules.logical.SubstituteSurrogateExpressions;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.SubstituteSurrogatePlans;
-import org.elasticsearch.xpack.esql.optimizer.rules.logical.SubstituteSurrogates;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.TranslateTimeSeriesAggregate;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.esql.rule.ParameterizedRuleExecutor;
@@ -137,11 +137,12 @@ public class LogicalPlanOptimizer extends ParameterizedRuleExecutor<LogicalPlan,
             // then extract nested aggs top-level
             new ReplaceAggregateAggExpressionWithEval(),
             // lastly replace surrogate functions
-            new SubstituteSurrogates(),
+            new SubstituteSurrogateAggregations(),
+            // translate metric aggregates after surrogate substitution and replace nested expressions with eval (again)
             new TranslateTimeSeriesAggregate(),
             new PruneUnusedIndexMode(),
             // after translating metric aggregates, we need to replace surrogate substitutions and nested expressions again.
-            new SubstituteSurrogates(),
+            new SubstituteSurrogateAggregations(),
             new ReplaceAggregateNestedExpressionWithEval(),
             // this one needs to be placed before ReplaceAliasingEvalWithProject, so that any potential aliasing eval (eval x = y)
             // is not replaced with a Project before the eval to be copied on the left hand side of an InlineJoin
@@ -150,7 +151,7 @@ public class LogicalPlanOptimizer extends ParameterizedRuleExecutor<LogicalPlan,
             new ReplaceTrivialTypeConversions(),
             new ReplaceAliasingEvalWithProject(),
             new SkipQueryOnEmptyMappings(),
-            new SubstituteSpatialSurrogates(),
+            new SubstituteSurrogateExpressions(),
             new ReplaceOrderByExpressionWithEval()
             // new NormalizeAggregate(), - waits on https://github.com/elastic/elasticsearch/issues/100634
         );

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/SubstituteSpatialSurrogates.java

@@ -1,30 +0,0 @@
-/*
- * 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.optimizer.rules.logical;
-
-import org.elasticsearch.xpack.esql.expression.function.scalar.spatial.SpatialRelatesFunction;
-import org.elasticsearch.xpack.esql.optimizer.LogicalOptimizerContext;
-
-/**
- * Currently this works similarly to SurrogateExpression, leaving the logic inside the expressions,
- * so each can decide for itself whether or not to change to a surrogate expression.
- * But what is actually being done is similar to LiteralsOnTheRight. We can consider in the future moving
- * this in either direction, reducing the number of rules, but for now,
- * it's a separate rule to reduce the risk of unintended interactions with other rules.
- */
-public final class SubstituteSpatialSurrogates extends OptimizerRules.OptimizerExpressionRule<SpatialRelatesFunction> {
-
-    public SubstituteSpatialSurrogates() {
-        super(OptimizerRules.TransformDirection.UP);
-    }
-
-    @Override
-    protected SpatialRelatesFunction rule(SpatialRelatesFunction function, LogicalOptimizerContext ctx) {
-        return function.surrogate();
-    }
-}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/SubstituteSurrogates.java → x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/SubstituteSurrogateAggregations.java

@@ -32,10 +32,8 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-public final class SubstituteSurrogates extends OptimizerRules.OptimizerRule<Aggregate> {
-    // TODO: currently this rule only works for aggregate functions (AVG)
-
-    public SubstituteSurrogates() {
+public final class SubstituteSurrogateAggregations extends OptimizerRules.OptimizerRule<Aggregate> {
+    public SubstituteSurrogateAggregations() {
         super(OptimizerRules.TransformDirection.UP);
     }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/SubstituteSurrogateExpressions.java

@@ -0,0 +1,40 @@
+/*
+ * 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.optimizer.rules.logical;
+
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.expression.SurrogateExpression;
+import org.elasticsearch.xpack.esql.optimizer.LogicalOptimizerContext;
+
+/**
+ * Replace {@link SurrogateExpression}s with their {@link SurrogateExpression#surrogate surrogates}.
+ */
+public final class SubstituteSurrogateExpressions extends OptimizerRules.OptimizerExpressionRule<Expression> {
+
+    public SubstituteSurrogateExpressions() {
+        super(OptimizerRules.TransformDirection.UP);
+    }
+
+    @Override
+    protected Expression rule(Expression e, LogicalOptimizerContext ctx) {
+        return rule(e);
+    }
+
+    /**
+     * Perform the actual substitution.
+     */
+    public static Expression rule(Expression e) {
+        if (e instanceof SurrogateExpression s) {
+            Expression surrogate = s.surrogate();
+            if (surrogate != null) {
+                return surrogate;
+            }
+        }
+        return e;
+    }
+}

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

@@ -45,8 +45,8 @@ import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToDatetim
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToDouble;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToGeoPoint;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToGeoShape;
-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.ToIpLeadingZerosRejected;
 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.ToTimeDuration;
@@ -122,7 +122,7 @@ public class EsqlDataTypeConverter {
         entry(GEO_POINT, ToGeoPoint::new),
         entry(GEO_SHAPE, ToGeoShape::new),
         entry(INTEGER, ToInteger::new),
-        entry(IP, ToIP::new),
+        entry(IP, ToIpLeadingZerosRejected::new),
         entry(LONG, ToLong::new),
         // ToRadians, typeless
         entry(KEYWORD, ToString::new),

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

@@ -13,10 +13,9 @@ import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.json.JsonXContent;
 import org.elasticsearch.xpack.esql.LoadMapping;
 import org.elasticsearch.xpack.esql.action.EsqlCapabilities;
-import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.expression.function.Function;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.expression.function.EsqlFunctionRegistry;
-import org.elasticsearch.xpack.esql.expression.function.FunctionDefinition;
 import org.elasticsearch.xpack.esql.index.EsIndex;
 import org.elasticsearch.xpack.esql.index.IndexResolution;
 import org.elasticsearch.xpack.esql.parser.EsqlParser;
@@ -100,9 +99,9 @@ public class ParsingTests extends ESTestCase {
                 LogicalPlan plan = parser.createStatement("ROW a = 1::" + nameOrAlias);
                 Row row = as(plan, Row.class);
                 assertThat(row.fields(), hasSize(1));
-                Expression functionCall = row.fields().get(0).child();
+                Function functionCall = (Function) row.fields().get(0).child();
                 assertThat(functionCall.dataType(), equalTo(expectedType));
-                report.field(nameOrAlias, functionName(registry, functionCall));
+                report.field(nameOrAlias, registry.functionName(functionCall.getClass()));
             }
             report.endObject();
         }
@@ -157,15 +156,6 @@ public class ParsingTests extends ESTestCase {
         );
     }
 
-    private String functionName(EsqlFunctionRegistry registry, Expression functionCall) {
-        for (FunctionDefinition def : registry.listFunctions()) {
-            if (functionCall.getClass().equals(def.clazz())) {
-                return def.name();
-            }
-        }
-        throw new IllegalArgumentException("can't find name for " + functionCall);
-    }
-
     private String error(String query) {
         ParsingException e = expectThrows(ParsingException.class, () -> defaultAnalyzer.analyze(parser.createStatement(query)));
         String message = e.getMessage();

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

@@ -42,6 +42,7 @@ import org.elasticsearch.xpack.esql.core.util.NumericUtils;
 import org.elasticsearch.xpack.esql.core.util.StringUtils;
 import org.elasticsearch.xpack.esql.evaluator.EvalMapper;
 import org.elasticsearch.xpack.esql.evaluator.mapper.EvaluatorMapper;
+import org.elasticsearch.xpack.esql.expression.SurrogateExpression;
 import org.elasticsearch.xpack.esql.expression.function.scalar.conditional.Greatest;
 import org.elasticsearch.xpack.esql.expression.function.scalar.nulls.Coalesce;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput;
@@ -484,13 +485,21 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
     }
 
     private Expression randomSerializeDeserialize(Expression expression) {
-        if (randomBoolean()) {
+        if (canSerialize() == false || randomBoolean()) {
             return expression;
         }
 
         return serializeDeserializeExpression(expression);
     }
 
+    /**
+     * The expression being tested be serialized? The <strong>vast</strong>
+     * majority of expressions can be serialized.
+     */
+    protected boolean canSerialize() {
+        return true;
+    }
+
     /**
      * Returns the expression after being serialized and deserialized.
      * <p>
@@ -545,6 +554,12 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
         if (e.foldable()) {
             e = new Literal(e.source(), e.fold(FoldContext.small()), e.dataType());
         }
+        if (e instanceof SurrogateExpression s) {
+            Expression surrogate = s.surrogate();
+            if (surrogate != null) {
+                e = surrogate;
+            }
+        }
         Layout.Builder builder = new Layout.Builder();
         buildLayout(builder, e);
         Expression.TypeResolution resolution = e.typeResolved();
@@ -705,6 +720,7 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
     }
 
     public void testSerializationOfSimple() {
+        assumeTrue("can't serialize function", canSerialize());
         assertSerialization(buildFieldExpression(testCase), testCase.getConfiguration());
     }
 

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

@@ -25,6 +25,7 @@ import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.FoldContext;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.util.NumericUtils;
+import org.elasticsearch.xpack.esql.expression.SurrogateExpression;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.FoldNull;
 import org.elasticsearch.xpack.esql.planner.PlannerUtils;
 import org.hamcrest.Matcher;
@@ -366,6 +367,12 @@ public abstract class AbstractScalarFunctionTestCase extends AbstractFunctionTes
             return;
         }
         assertFalse("expected resolved", expression.typeResolved().unresolved());
+        if (expression instanceof SurrogateExpression s) {
+            Expression surrogate = s.surrogate();
+            if (surrogate != null) {
+                expression = surrogate;
+            }
+        }
         Expression nullOptimized = new FoldNull().rule(expression, unboundLogicalOptimizerContext());
         assertThat(nullOptimized.dataType(), equalTo(testCase.expectedType()));
         assertTrue(nullOptimized.foldable());

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

@@ -1,87 +0,0 @@
-/*
- * 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 com.carrotsearch.randomizedtesting.annotations.Name;
-import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
-
-import org.apache.lucene.util.BytesRef;
-import org.elasticsearch.common.network.NetworkAddress;
-import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.xpack.esql.core.expression.Expression;
-import org.elasticsearch.xpack.esql.core.tree.Source;
-import org.elasticsearch.xpack.esql.core.type.DataType;
-import org.elasticsearch.xpack.esql.expression.function.AbstractScalarFunctionTestCase;
-import org.elasticsearch.xpack.esql.expression.function.TestCaseSupplier;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.function.Supplier;
-
-import static java.util.Collections.emptyList;
-import static org.elasticsearch.xpack.esql.core.util.StringUtils.parseIP;
-
-public class ToIPTests extends AbstractScalarFunctionTestCase {
-    public ToIPTests(@Name("TestCase") Supplier<TestCaseSupplier.TestCase> testCaseSupplier) {
-        this.testCase = testCaseSupplier.get();
-    }
-
-    @ParametersFactory
-    public static Iterable<Object[]> parameters() {
-        String read = "Attribute[channel=0]";
-        String stringEvaluator = "ParseIpLeadingZerosRejectedEvaluator[string=" + read + "]";
-        List<TestCaseSupplier> suppliers = new ArrayList<>();
-
-        // convert from IP to IP
-        TestCaseSupplier.forUnaryIp(suppliers, read, DataType.IP, v -> v, List.of());
-
-        // convert random string (i.e. not an IP representation) to IP `null`, with warnings.
-        TestCaseSupplier.forUnaryStrings(
-            suppliers,
-            stringEvaluator,
-            DataType.IP,
-            bytesRef -> null,
-            bytesRef -> List.of(
-                "Line 1:1: evaluation of [source] failed, treating result as null. Only first 20 failures recorded.",
-                "Line 1:1: java.lang.IllegalArgumentException: '" + bytesRef.utf8ToString() + "' is not an IP string literal."
-            )
-        );
-
-        // convert valid IPs shaped as strings
-        TestCaseSupplier.unary(
-            suppliers,
-            stringEvaluator,
-            validIPsAsStrings(),
-            DataType.IP,
-            bytesRef -> parseIP(((BytesRef) bytesRef).utf8ToString()),
-            emptyList()
-        );
-        return parameterSuppliersFromTypedDataWithDefaultChecksNoErrors(true, suppliers);
-    }
-
-    @Override
-    protected Expression build(Source source, List<Expression> args) {
-        return new ToIP(source, args.get(0));
-    }
-
-    private static List<TestCaseSupplier.TypedDataSupplier> validIPsAsStrings() {
-        return List.of(
-            new TestCaseSupplier.TypedDataSupplier("<127.0.0.1 ip>", () -> new BytesRef("127.0.0.1"), DataType.KEYWORD),
-            new TestCaseSupplier.TypedDataSupplier(
-                "<ipv4>",
-                () -> new BytesRef(NetworkAddress.format(ESTestCase.randomIp(true))),
-                DataType.KEYWORD
-            ),
-            new TestCaseSupplier.TypedDataSupplier(
-                "<ipv6>",
-                () -> new BytesRef(NetworkAddress.format(ESTestCase.randomIp(false))),
-                DataType.TEXT
-            )
-        );
-    }
-}

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToIPErrorTests.java → x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToIpErrorTests.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.expression.function.scalar.convert;
 
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
@@ -19,15 +20,15 @@ import java.util.Set;
 
 import static org.hamcrest.Matchers.equalTo;
 
-public class ToIPErrorTests extends ErrorsForCasesWithoutExamplesTestCase {
+public class ToIpErrorTests extends ErrorsForCasesWithoutExamplesTestCase {
     @Override
     protected List<TestCaseSupplier> cases() {
-        return paramsToSuppliers(ToIPTests.parameters());
+        return Iterators.toList(Iterators.map(ToIpTests.parameters().iterator(), p -> (TestCaseSupplier) p[0]));
     }
 
     @Override
     protected Expression build(Source source, List<Expression> args) {
-        return new ToIP(source, args.get(0));
+        return new ToIp(source, args.getFirst(), null);
     }
 
     @Override

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

@@ -0,0 +1,19 @@
+/*
+ * 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.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.expression.AbstractUnaryScalarSerializationTests;
+
+public class ToIpLeadingZerosDecimalSerializationTests extends AbstractUnaryScalarSerializationTests<ToIpLeadingZerosDecimal> {
+    @Override
+    protected ToIpLeadingZerosDecimal create(Source source, Expression child) {
+        return new ToIpLeadingZerosDecimal(source, child);
+    }
+}

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToIPSerializationTests.java → x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToIpLeadingZerosOctalSerializationTests.java

@@ -11,9 +11,9 @@ import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.expression.AbstractUnaryScalarSerializationTests;
 
-public class ToIPSerializationTests extends AbstractUnaryScalarSerializationTests<ToIP> {
+public class ToIpLeadingZerosOctalSerializationTests extends AbstractUnaryScalarSerializationTests<ToIpLeadingZerosOctal> {
     @Override
-    protected ToIP create(Source source, Expression child) {
-        return new ToIP(source, child);
+    protected ToIpLeadingZerosOctal create(Source source, Expression child) {
+        return new ToIpLeadingZerosOctal(source, child);
     }
 }

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

@@ -0,0 +1,19 @@
+/*
+ * 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.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.expression.AbstractUnaryScalarSerializationTests;
+
+public class ToIpLeadingZerosRejectedSerializationTests extends AbstractUnaryScalarSerializationTests<ToIpLeadingZerosRejected> {
+    @Override
+    protected ToIpLeadingZerosRejected create(Source source, Expression child) {
+        return new ToIpLeadingZerosRejected(source, child);
+    }
+}

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

@@ -0,0 +1,197 @@
+/*
+ * 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 com.carrotsearch.randomizedtesting.annotations.Name;
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.network.NetworkAddress;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.expression.Literal;
+import org.elasticsearch.xpack.esql.core.expression.MapExpression;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.expression.function.AbstractScalarFunctionTestCase;
+import org.elasticsearch.xpack.esql.expression.function.TestCaseSupplier;
+import org.elasticsearch.xpack.esql.type.EsqlDataTypeConverter;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.function.Supplier;
+
+import static java.util.Collections.emptyList;
+import static org.elasticsearch.xpack.esql.core.util.StringUtils.parseIP;
+import static org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToIp.LeadingZeros.DECIMAL;
+import static org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToIp.LeadingZeros.OCTAL;
+import static org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToIp.LeadingZeros.REJECT;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.nullValue;
+
+public class ToIpTests extends AbstractScalarFunctionTestCase {
+    private final ToIp.LeadingZeros leadingZeros;
+
+    public ToIpTests(
+        @Name("TestCase") Supplier<TestCaseSupplier.TestCase> testCaseSupplier,
+        @Name("leading_zeros") ToIp.LeadingZeros leadingZeros
+    ) {
+        this.testCase = testCaseSupplier.get();
+        this.leadingZeros = leadingZeros;
+    }
+
+    @ParametersFactory
+    public static Iterable<Object[]> parameters() {
+        List<Object[]> parameters = new ArrayList<>();
+        for (ToIp.LeadingZeros leadingZeros : new ToIp.LeadingZeros[] { null, REJECT, OCTAL, DECIMAL }) {
+            List<TestCaseSupplier> suppliers = new ArrayList<>();
+            // convert from IP to IP
+            TestCaseSupplier.forUnaryIp(suppliers, readEvaluator(), DataType.IP, v -> v, List.of());
+
+            // convert random string (i.e. not an IP representation) to IP `null`, with warnings.
+            TestCaseSupplier.forUnaryStrings(
+                suppliers,
+                stringEvaluator(leadingZeros),
+                DataType.IP,
+                bytesRef -> null,
+                bytesRef -> List.of(
+                    "Line 1:1: evaluation of [source] failed, treating result as null. Only first 20 failures recorded.",
+                    "Line 1:1: java.lang.IllegalArgumentException: '" + bytesRef.utf8ToString() + "' is not an IP string literal."
+                )
+            );
+
+            // convert valid IPs shaped as strings
+            TestCaseSupplier.unary(
+                suppliers,
+                stringEvaluator(leadingZeros),
+                validIPsAsStrings(),
+                DataType.IP,
+                bytesRef -> parseIP(((BytesRef) bytesRef).utf8ToString()),
+                emptyList()
+            );
+            suppliers = anyNullIsNull(true, randomizeBytesRefsOffset(suppliers));
+            for (TestCaseSupplier supplier : suppliers) {
+                parameters.add(new Object[] { supplier, leadingZeros });
+            }
+        }
+
+        parameters.add(new Object[] { exampleRejectingLeadingZeros(stringEvaluator(null)), null });
+        parameters.add(new Object[] { exampleRejectingLeadingZeros(stringEvaluator(REJECT)), REJECT });
+        parameters.add(new Object[] { exampleParsingLeadingZerosAsDecimal(stringEvaluator(DECIMAL)), DECIMAL });
+        parameters.add(new Object[] { exampleParsingLeadingZerosAsOctal(stringEvaluator(OCTAL)), OCTAL });
+        return parameters;
+    }
+
+    private static TestCaseSupplier exampleRejectingLeadingZeros(String stringEvaluator) {
+        return new TestCaseSupplier("<ip> with leading 0s", List.of(DataType.KEYWORD), () -> {
+            BytesRef withLeadingZeros = new BytesRef(randomIpWithLeadingZeros());
+            return new TestCaseSupplier.TestCase(
+                List.of(new TestCaseSupplier.TypedData(withLeadingZeros, DataType.KEYWORD, "ip")),
+                stringEvaluator,
+                DataType.IP,
+                nullValue()
+            ).withWarning("Line 1:1: evaluation of [source] failed, treating result as null. Only first 20 failures recorded.")
+                .withWarning(
+                    "Line 1:1: java.lang.IllegalArgumentException: '" + withLeadingZeros.utf8ToString() + "' is not an IP string literal."
+                );
+        });
+    }
+
+    private static TestCaseSupplier exampleParsingLeadingZerosAsDecimal(String stringEvaluator) {
+        return new TestCaseSupplier("<ip> with leading 0s", List.of(DataType.KEYWORD), () -> {
+            String ip = randomIpWithLeadingZeros();
+            BytesRef withLeadingZeros = new BytesRef(ip);
+            String withoutLeadingZeros = ParseIpTests.leadingZerosAreDecimalToIp(ip);
+            return new TestCaseSupplier.TestCase(
+                List.of(new TestCaseSupplier.TypedData(withLeadingZeros, DataType.KEYWORD, "ip")),
+                stringEvaluator,
+                DataType.IP,
+                equalTo(EsqlDataTypeConverter.stringToIP(withoutLeadingZeros))
+            );
+        });
+    }
+
+    private static TestCaseSupplier exampleParsingLeadingZerosAsOctal(String stringEvaluator) {
+        return new TestCaseSupplier("<ip> with leading 0s", List.of(DataType.KEYWORD), () -> {
+            String ip = randomIpWithLeadingZerosOctal();
+            BytesRef withLeadingZeros = new BytesRef(ip);
+            String withoutLeadingZeros = ParseIpTests.leadingZerosAreOctalToIp(ip);
+            return new TestCaseSupplier.TestCase(
+                List.of(new TestCaseSupplier.TypedData(withLeadingZeros, DataType.KEYWORD, "ip")),
+                stringEvaluator,
+                DataType.IP,
+                equalTo(EsqlDataTypeConverter.stringToIP(withoutLeadingZeros))
+            );
+        });
+    }
+
+    @Override
+    protected Expression build(Source source, List<Expression> args) {
+        return new ToIp(source, args.getFirst(), options());
+    }
+
+    private MapExpression options() {
+        if (leadingZeros == null) {
+            return null;
+        }
+        return new MapExpression(
+            Source.EMPTY,
+            List.of(
+                new Literal(Source.EMPTY, "leading_zeros", DataType.KEYWORD),
+                new Literal(Source.EMPTY, leadingZeros.toString().toLowerCase(Locale.ROOT), DataType.KEYWORD)
+            )
+        );
+    }
+
+    private static List<TestCaseSupplier.TypedDataSupplier> validIPsAsStrings() {
+        return List.of(
+            new TestCaseSupplier.TypedDataSupplier("<127.0.0.1 ip>", () -> new BytesRef("127.0.0.1"), DataType.KEYWORD),
+            new TestCaseSupplier.TypedDataSupplier(
+                "<ipv4>",
+                () -> new BytesRef(NetworkAddress.format(ESTestCase.randomIp(true))),
+                DataType.KEYWORD
+            ),
+            new TestCaseSupplier.TypedDataSupplier(
+                "<ipv6>",
+                () -> new BytesRef(NetworkAddress.format(ESTestCase.randomIp(false))),
+                DataType.TEXT
+            )
+        );
+    }
+
+    private static String randomIpWithLeadingZeros() {
+        return randomValueOtherThanMany((String str) -> false == (str.startsWith("0") || str.contains(".0")), () -> {
+            byte[] addr = randomIp(true).getAddress();
+            return String.format(Locale.ROOT, "%03d.%03d.%03d.%03d", addr[0] & 0xff, addr[1] & 0xff, addr[2] & 0xff, addr[3] & 0xff);
+        });
+    }
+
+    private static String randomIpWithLeadingZerosOctal() {
+        byte[] addr = randomIp(true).getAddress();
+        return String.format(Locale.ROOT, "0%o.0%o.0%o.0%o", addr[0] & 0xff, addr[1] & 0xff, addr[2] & 0xff, addr[3] & 0xff);
+    }
+
+    private static String readEvaluator() {
+        return "Attribute[channel=0]";
+    }
+
+    private static String stringEvaluator(ToIp.LeadingZeros leadingZeros) {
+        return switch (leadingZeros) {
+            case null -> "ParseIpLeadingZerosRejectedEvaluator";
+            case REJECT -> "ParseIpLeadingZerosRejectedEvaluator";
+            case DECIMAL -> "ParseIpLeadingZerosAreDecimalEvaluator";
+            case OCTAL -> "ParseIpLeadingZerosAreOctalEvaluator";
+        } + "[string=" + readEvaluator() + "]";
+    }
+
+    @Override
+    protected boolean canSerialize() {
+        return false;
+    }
+}

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/type/MultiTypeEsFieldTests.java

@@ -24,8 +24,8 @@ import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToDatetim
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToDouble;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToGeoPoint;
 import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToGeoShape;
-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.ToIpLeadingZerosRejected;
 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;
@@ -156,7 +156,7 @@ public class MultiTypeEsFieldTests extends AbstractWireTestCase<MultiTypeEsField
                 case DOUBLE, FLOAT -> new ToDouble(Source.EMPTY, fromField);
                 case INTEGER -> new ToInteger(Source.EMPTY, fromField);
                 case LONG -> new ToLong(Source.EMPTY, fromField);
-                case IP -> new ToIP(Source.EMPTY, fromField);
+                case IP -> new ToIpLeadingZerosRejected(Source.EMPTY, fromField);
                 case KEYWORD -> new ToString(Source.EMPTY, fromField);
                 case GEO_POINT -> new ToGeoPoint(Source.EMPTY, fromField);
                 case GEO_SHAPE -> new ToGeoShape(Source.EMPTY, fromField);