[ES|QL] String literal implicit casting

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

@@ -510,6 +510,17 @@ date1:date          |         date2:date          |    dd_ms:integer
 // end::docsDateDiff-result[]
 ;
 
+evalDateDiffString
+required_feature: esql.string_literal_auto_casting
+
+ROW date1 = TO_DATETIME("2023-12-02T11:00:00.000Z")
+| EVAL dd_ms = DATE_DIFF("microseconds", date1, "2023-12-02T11:00:00.001Z")
+;
+
+date1:date               | dd_ms:integer
+2023-12-02T11:00:00.000Z | 1000
+;
+
 evalDateParseWithSimpleDate
 row a = "2023-02-01" | eval b = date_parse("yyyy-MM-dd", a) | keep b;
 
@@ -1085,6 +1096,17 @@ date:date                 | year:long
 // end::dateExtract-result[]
 ;
 
+dateExtractString
+required_feature: esql.string_literal_auto_casting
+
+ROW date = DATE_PARSE("yyyy-MM-dd", "2022-05-06")
+| EVAL year = DATE_EXTRACT("year", "2022-05-06")
+;
+
+date:date                 | year:long
+2022-05-06T00:00:00.000Z  | 2022
+;
+
 docsDateExtractBusinessHours
 // tag::docsDateExtractBusinessHours[]
 FROM sample_data
@@ -1115,6 +1137,17 @@ Anneke         |Preusig        |1989-06-02T00:00:00.000Z|1989-06-02
 // end::docsDateFormat-result[]
 ;
 
+evalDateFormatString
+required_feature: esql.string_literal_auto_casting
+
+ROW a = 1
+| EVAL df = DATE_FORMAT("YYYY-MM-dd", "1989-06-02T00:00:00.000Z")
+;
+
+a:integer | df:keyword
+1         | 1989-06-02
+;
+
 docsDateTrunc
 // tag::docsDateTrunc[]
 FROM employees
@@ -1133,6 +1166,17 @@ Anneke         |Preusig        |1989-06-02T00:00:00.000Z|1989-01-01T00:00:00.000
 // end::docsDateTrunc-result[]
 ;
 
+evalDateTruncString
+required_feature: esql.string_literal_auto_casting
+
+ROW a = 1
+| EVAL year_hired = DATE_TRUNC(1 year, "1991-06-26T00:00:00.000Z")
+;
+
+a:integer | year_hired:date
+1         | 1991-01-01T00:00:00.000Z
+;
+
 docsDateTruncHistogram
 // tag::docsDateTruncHistogram[]
 FROM employees

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

@@ -1427,3 +1427,68 @@ Amabile        |Gomatam        |2.09           |2.09
 Anneke         |Preusig        |1.56           |1.56     
 //end::abs-employees-result[]
 ;
+
+evalAbsString
+required_feature: esql.string_literal_auto_casting
+
+ROW number = -1.0
+| EVAL abs_number = ABS("10.0")
+;
+
+number:double  | abs_number:double
+-1.0           | 10.0
+;
+
+arithmeticOperationWithString
+required_feature: esql.string_literal_auto_casting
+
+from employees
+| eval s1 = salary + "10000",  s2 = height * "2", s3 = avg_worked_seconds / "2", s4 = languages - "1"
+| sort emp_no
+| keep emp_no, salary, s1, height, s2, avg_worked_seconds, s3, languages, s4
+| limit 2;
+
+emp_no:integer | salary:integer | s1:integer | height:double | s2:double | avg_worked_seconds:long | s3:long   | languages:integer | s4:integer
+10001          | 57305          | 67305      | 2.03          | 4.06      | 268728049               | 134364024 | 2                 | 1
+10002          | 56371          | 66371      | 2.08          | 4.16      | 328922887               | 164461443 | 5                 | 4
+;
+
+arithmeticOperationNestedWithString
+required_feature: esql.string_literal_auto_casting
+
+from employees
+| eval x = languages + "1", y = x * 2
+| sort emp_no
+| keep emp_no, languages, x, y
+| limit 2;
+
+emp_no: integer | languages:integer | x:integer | y:integer
+10001           | 2                 | 3         | 6
+10002           | 5                 | 6         | 12
+;
+
+functionUnderArithmeticOperationAggString
+required_feature: esql.string_literal_auto_casting
+
+ROW a = 1
+| eval x = date_trunc(1 month, "2024-11-22") + 2 days, y = x + 3 days
+| stats count() by y
+;
+
+count():long | y:date
+1            | 2024-11-06T00:00:00.000Z
+;
+
+functionUnderArithmeticOperationString
+required_feature: esql.string_literal_auto_casting
+
+from employees
+| eval x = date_trunc(1 month, "2024-11-22") + 2 days, y = x + 3 days
+| sort emp_no
+| keep emp_no, x, y
+| limit 2;
+
+emp_no: integer | x:date     | y:date
+10001           | 2024-11-03 | 2024-11-06
+10002           | 2024-11-03 | 2024-11-06
+;

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

@@ -7,7 +7,7 @@ synopsis:keyword
 "double asin(number:double|integer|long|unsigned_long)"
 "double atan(number:double|integer|long|unsigned_long)"
 "double atan2(y_coordinate:double|integer|long|unsigned_long, x_coordinate:double|integer|long|unsigned_long)"
-"double|date auto_bucket(field:integer|long|double|date, buckets:integer, from:integer|long|double|date|string, to:integer|long|double|date|string)"
+"double|date auto_bucket(field:integer|long|double|date, buckets:integer, from:integer|long|double|date, to:integer|long|double|date)"
 "double avg(number:double|integer|long)"
 "boolean|cartesian_point|date|double|geo_point|integer|ip|keyword|long|text|unsigned_long|version case(condition:boolean, trueValue...:boolean|cartesian_point|date|double|geo_point|integer|ip|keyword|long|text|unsigned_long|version)"
 "double|integer|long|unsigned_long ceil(number:double|integer|long|unsigned_long)"
@@ -117,7 +117,7 @@ acos          |number                              |"double|integer|long|unsigne
 asin          |number                              |"double|integer|long|unsigned_long"                                                                                               |Number between -1 and 1. If `null`, the function returns `null`.
 atan          |number                              |"double|integer|long|unsigned_long"                                                                                               |Numeric expression. If `null`, the function returns `null`.
 atan2         |[y_coordinate, x_coordinate]        |["double|integer|long|unsigned_long", "double|integer|long|unsigned_long"]                                                        |[y coordinate. If `null`\, the function returns `null`., x coordinate. If `null`\, the function returns `null`.]
-auto_bucket   |[field, buckets, from, to]          |["integer|long|double|date", integer, "integer|long|double|date|string", "integer|long|double|date|string"]                       |["", "", "", ""]
+auto_bucket   |[field, buckets, from, to]          |["integer|long|double|date", integer, "integer|long|double|date", "integer|long|double|date"]                       |["", "", "", ""]
 avg           |number                              |"double|integer|long"                                                                                                             |[""]
 case          |[condition, trueValue]              |[boolean, "boolean|cartesian_point|date|double|geo_point|integer|ip|keyword|long|text|unsigned_long|version"]                     |["", ""]
 ceil          |number                              |"double|integer|long|unsigned_long"                                                                                               |Numeric expression. If `null`, the function returns `null`.

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

@@ -70,6 +70,29 @@ emp_no:integer | last_name:keyword | gender:keyword | f_l:boolean
 10010 | Piveteau  | null | null
 ;
 
+stringCast
+required_feature: esql.string_literal_auto_casting
+
+ROW a = 1 | eval ss = substring("abcd", "2"), l = left("abcd", "2"), r = right("abcd", "2");
+
+a:integer | ss:keyword | l:keyword | r:keyword
+1         | bcd        | ab        | cd
+;
+
+stringCastEmp
+required_feature: esql.string_literal_auto_casting
+
+from employees
+| eval ss = substring(first_name, "2")
+| sort emp_no
+| keep emp_no, first_name, ss
+| limit 2;
+
+emp_no: integer | first_name:keyword | ss:keyword
+10001           | Georgi             | eorgi
+10002           | Bezalel            | ezalel
+;
+
 substring
 from employees | where emp_no <= 10010 | eval f_l = substring(last_name, 3) | keep emp_no, last_name, f_l;
 ignoreOrder:true
@@ -748,6 +771,16 @@ emp_no:integer | job_positions:keyword
 10005          | null                                                                   | null                                                                   | null
 ;
 
+mvSliceCast
+required_feature: esql.string_literal_auto_casting
+
+ROW a = ["1", "2", "3", "4"]
+| eval a1 = mv_slice(a, "0", "1");
+
+a:keyword            | a1:keyword
+["1", "2", "3", "4"] | ["1", "2"]
+;
+
 mvSliceEmp
 required_feature: esql.mv_sort
 

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

@@ -8,12 +8,16 @@
 package org.elasticsearch.xpack.esql.analysis;
 
 import org.elasticsearch.common.logging.HeaderWarning;
+import org.elasticsearch.common.logging.LoggerMessageFormat;
 import org.elasticsearch.xpack.core.enrich.EnrichPolicy;
 import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
 import org.elasticsearch.xpack.esql.VerificationException;
 import org.elasticsearch.xpack.esql.expression.NamedExpressions;
 import org.elasticsearch.xpack.esql.expression.UnresolvedNamePattern;
+import org.elasticsearch.xpack.esql.expression.function.EsqlFunctionRegistry;
 import org.elasticsearch.xpack.esql.expression.function.UnsupportedAttribute;
+import org.elasticsearch.xpack.esql.expression.function.scalar.EsqlScalarFunction;
+import org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic.EsqlArithmeticOperation;
 import org.elasticsearch.xpack.esql.plan.logical.Drop;
 import org.elasticsearch.xpack.esql.plan.logical.Enrich;
 import org.elasticsearch.xpack.esql.plan.logical.EsqlAggregate;
@@ -24,6 +28,7 @@
 import org.elasticsearch.xpack.esql.plan.logical.Rename;
 import org.elasticsearch.xpack.esql.plan.logical.local.EsqlProject;
 import org.elasticsearch.xpack.esql.stats.FeatureMetric;
+import org.elasticsearch.xpack.esql.type.EsqlDataTypeConverter;
 import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
 import org.elasticsearch.xpack.ql.analyzer.AnalyzerRules;
 import org.elasticsearch.xpack.ql.analyzer.AnalyzerRules.BaseAnalyzerRule;
@@ -46,6 +51,7 @@
 import org.elasticsearch.xpack.ql.expression.function.FunctionDefinition;
 import org.elasticsearch.xpack.ql.expression.function.FunctionRegistry;
 import org.elasticsearch.xpack.ql.expression.function.UnresolvedFunction;
+import org.elasticsearch.xpack.ql.expression.function.scalar.ScalarFunction;
 import org.elasticsearch.xpack.ql.expression.predicate.operator.comparison.BinaryComparison;
 import org.elasticsearch.xpack.ql.index.EsIndex;
 import org.elasticsearch.xpack.ql.plan.TableIdentifier;
@@ -110,7 +116,14 @@ public class Analyzer extends ParameterizedRuleExecutor<LogicalPlan, AnalyzerCon
     private static final Iterable<RuleExecutor.Batch<LogicalPlan>> rules;
 
     static {
-        var resolution = new Batch<>("Resolution", new ResolveTable(), new ResolveEnrich(), new ResolveFunctions(), new ResolveRefs());
+        var resolution = new Batch<>(
+            "Resolution",
+            new ResolveTable(),
+            new ResolveEnrich(),
+            new ResolveFunctions(),
+            new ResolveRefs(),
+            new ImplicitCasting()
+        );
         var finish = new Batch<>("Finish Analysis", Limiter.ONCE, new AddImplicitLimit(), new PromoteStringsInDateComparisons());
         rules = List.of(resolution, finish);
     }
@@ -780,13 +793,13 @@ private static Expression promote(BinaryComparison cmp) {
             var right = cmp.right();
             boolean modified = false;
             if (left.dataType() == DATETIME) {
-                if (right.dataType() == KEYWORD && right.foldable()) {
+                if (right.dataType() == KEYWORD && right.foldable() && ((right instanceof EsqlScalarFunction) == false)) {
                     right = stringToDate(right);
                     modified = true;
                 }
             } else {
                 if (right.dataType() == DATETIME) {
-                    if (left.dataType() == KEYWORD && left.foldable()) {
+                    if (left.dataType() == KEYWORD && left.foldable() && ((left instanceof EsqlScalarFunction) == false)) {
                         left = stringToDate(left);
                         modified = true;
                     }
@@ -825,4 +838,118 @@ private BitSet gatherPreAnalysisMetrics(LogicalPlan plan, BitSet b) {
         plan.forEachDown(p -> FeatureMetric.set(p, b));
         return b;
     }
+
+    private static class ImplicitCasting extends ParameterizedAnalyzerRule<LogicalPlan, AnalyzerContext> {
+        @Override
+        protected LogicalPlan rule(LogicalPlan plan, AnalyzerContext context) {
+            return plan.transformExpressionsUp(ScalarFunction.class, e -> ImplicitCasting.cast(e, context.functionRegistry()));
+        }
+
+        private static Expression cast(ScalarFunction f, FunctionRegistry registry) {
+            if (f instanceof EsqlScalarFunction esf) {
+                return processScalarFunction(esf, registry);
+            }
+
+            if (f instanceof EsqlArithmeticOperation eao) {
+                return processArithmeticOperation(eao);
+            }
+
+            return f;
+        }
+
+        private static List<DataType> getTargetDataType(String name, FunctionRegistry registry) {
+            List<FunctionDefinition> defs = registry.listFunctions()
+                .stream()
+                .filter(f -> f.clazz().getSimpleName().equalsIgnoreCase(name))
+                .toList();
+            if (defs.isEmpty()) {
+                return new ArrayList<>();
+            }
+            EsqlFunctionRegistry.FunctionDescription signature = EsqlFunctionRegistry.description(defs.get(0));
+            return signature.args().stream().map(EsqlFunctionRegistry.ArgSignature::targetDataType).collect(Collectors.toList());
+        }
+
+        private static Expression processScalarFunction(EsqlScalarFunction f, FunctionRegistry registry) {
+            List<Expression> args = f.arguments();
+            List<DataType> targetDataTypes = getTargetDataType(f.functionName(), registry);
+            if (targetDataTypes.isEmpty()) {
+                return f;
+            }
+            List<Expression> newChildren = new ArrayList<>(args.size());
+            boolean childrenChanged = false;
+            DataType targetDataType = DataTypes.NULL;
+            Expression arg;
+            for (int i = 0; i < args.size(); i++) {
+                arg = args.get(i);
+                if (arg.resolved() && arg.dataType() == KEYWORD && arg.foldable() && ((arg instanceof EsqlScalarFunction) == false)) {
+                    if (i < targetDataTypes.size()) {
+                        targetDataType = targetDataTypes.get(i);
+                    }
+                    if (targetDataType != DataTypes.NULL && targetDataType != DataTypes.UNSUPPORTED) {
+                        Expression e = castStringLiteral(arg, targetDataType);
+                        childrenChanged = true;
+                        newChildren.add(e);
+                        continue;
+                    }
+                }
+                newChildren.add(args.get(i));
+            }
+            return childrenChanged ? f.replaceChildren(newChildren) : f;
+        }
+
+        private static Expression processArithmeticOperation(EsqlArithmeticOperation o) {
+            Expression left = o.left();
+            Expression right = o.right();
+            if (left.resolved() == false || right.resolved() == false) {
+                return o;
+            }
+            List<Expression> newChildren = new ArrayList<>(2);
+            boolean childrenChanged = false;
+            DataType targetDataType = DataTypes.NULL;
+            Expression from = Literal.NULL;
+
+            if (left.dataType() == KEYWORD
+                && left.foldable()
+                && (right.dataType().isNumeric() || right.dataType() == DATETIME)
+                && ((left instanceof EsqlScalarFunction) == false)) {
+                targetDataType = right.dataType();
+                from = left;
+            }
+            if (right.dataType() == KEYWORD
+                && right.foldable()
+                && (left.dataType().isNumeric() || left.dataType() == DATETIME)
+                && ((right instanceof EsqlScalarFunction) == false)) {
+                targetDataType = left.dataType();
+                from = right;
+            }
+            if (from != Literal.NULL) {
+                Expression e = castStringLiteral(from, targetDataType);
+                newChildren.add(from == left ? e : left);
+                newChildren.add(from == right ? e : right);
+                childrenChanged = true;
+            }
+            return childrenChanged ? o.replaceChildren(newChildren) : o;
+        }
+
+        public static Expression castStringLiteral(Expression from, DataType target) {
+            assert from.foldable();
+            try {
+                Object to = EsqlDataTypeConverter.convert(from.fold(), target);
+                return new Literal(from.source(), to, target);
+            } catch (Exception e) {
+                String message = LoggerMessageFormat.format(
+                    "Cannot convert string [{}] to [{}], error [{}]",
+                    from.fold(),
+                    target,
+                    e.getMessage()
+                );
+                return new UnsupportedAttribute(
+                    from.source(),
+                    String.valueOf(from.fold()),
+                    new UnsupportedEsField(String.valueOf(from.fold()), from.dataType().typeName()),
+                    message
+                );
+            }
+        }
+    }
 }