diff --git a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/Literal.java b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/Literal.java
index afe616489d81d..1bfb96a1b0f9d 100644
--- a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/Literal.java
+++ b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/Literal.java
@@ -11,18 +11,24 @@
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.xpack.esql.core.QlIllegalArgumentException;
 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.core.util.PlanStreamInput;
+import org.elasticsearch.xpack.versionfield.Version;
 
 import java.io.IOException;
+import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
 
 import static org.elasticsearch.xpack.esql.core.type.DataType.CARTESIAN_POINT;
 import static org.elasticsearch.xpack.esql.core.type.DataType.GEO_POINT;
+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.core.type.DataType.VERSION;
 import static org.elasticsearch.xpack.esql.core.util.SpatialCoordinateTypes.CARTESIAN;
 import static org.elasticsearch.xpack.esql.core.util.SpatialCoordinateTypes.GEO;
 
@@ -45,10 +51,25 @@ public class Literal extends LeafExpression {
 
     public Literal(Source source, Object value, DataType dataType) {
         super(source);
+        assert noPlainStrings(value, dataType);
         this.dataType = dataType;
         this.value = value;
     }
 
+    private boolean noPlainStrings(Object value, DataType dataType) {
+        if (dataType == KEYWORD || dataType == TEXT || dataType == VERSION) {
+            if (value == null) {
+                return true;
+            }
+            return switch (value) {
+                case String s -> false;
+                case Collection<?> c -> c.stream().allMatch(x -> noPlainStrings(x, dataType));
+                default -> true;
+            };
+        }
+        return true;
+    }
+
     private static Literal readFrom(StreamInput in) throws IOException {
         Source source = Source.readFrom((StreamInput & PlanStreamInput) in);
         Object value = in.readGenericValue();
@@ -122,7 +143,21 @@ public boolean equals(Object obj) {
 
     @Override
     public String toString() {
-        String str = String.valueOf(value);
+        String str;
+        if (dataType == KEYWORD || dataType == TEXT) {
+            str = BytesRefs.toString(value);
+        } else if (dataType == VERSION && value instanceof BytesRef br) {
+            str = new Version(br).toString();
+            // TODO review how we manage IPs: https://github.com/elastic/elasticsearch/issues/129605
+            // } else if (dataType == IP && value instanceof BytesRef ip) {
+            // str = DocValueFormat.IP.format(ip);
+        } else {
+            str = String.valueOf(value);
+        }
+
+        if (str == null) {
+            str = "null";
+        }
         if (str.length() > 500) {
             return str.substring(0, 500) + "...";
         }
@@ -154,6 +189,10 @@ public static Literal of(Expression source, Object value) {
         return new Literal(source.source(), value, source.dataType());
     }
 
+    public static Literal keyword(Source source, String literal) {
+        return new Literal(source, BytesRefs.toBytesRef(literal), KEYWORD);
+    }
+
     /**
      * Not all literal values are currently supported in StreamInput/StreamOutput as generic values.
      * This mapper allows for addition of new and interesting values without (yet) adding to StreamInput/Output.
diff --git a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/MapExpression.java b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/MapExpression.java
index 861ecc4ca0368..24736ac3a2514 100644
--- a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/MapExpression.java
+++ b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/MapExpression.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
@@ -39,7 +40,7 @@ public class MapExpression extends Expression {
 
     private final Map<Expression, Expression> map;
 
-    private final Map<Object, Expression> keyFoldedMap;
+    private final Map<String, Expression> keyFoldedMap;
 
     public MapExpression(Source source, List<Expression> entries) {
         super(source, entries);
@@ -54,7 +55,7 @@ public MapExpression(Source source, List<Expression> entries) {
             entryExpressions.add(new EntryExpression(key.source(), key, value));
             map.put(key, value);
             if (key instanceof Literal l) {
-                this.keyFoldedMap.put(l.value(), value);
+                this.keyFoldedMap.put(BytesRefs.toString(l.value()), value);
             }
         }
     }
@@ -95,7 +96,7 @@ public Map<Expression, Expression> map() {
         return map;
     }
 
-    public Map<Object, Expression> keyFoldedMap() {
+    public Map<String, Expression> keyFoldedMap() {
         return keyFoldedMap;
     }
 
diff --git a/x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/expression/LiteralTests.java b/x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/expression/LiteralTests.java
index a628916e67746..794d666ab744d 100644
--- a/x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/expression/LiteralTests.java
+++ b/x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/expression/LiteralTests.java
@@ -8,6 +8,7 @@
 
 import joptsimple.internal.Strings;
 
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.esql.core.InvalidArgumentException;
 import org.elasticsearch.xpack.esql.core.tree.AbstractNodeTestCase;
@@ -20,7 +21,6 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
-import java.util.Objects;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -62,7 +62,7 @@ static class ValueAndCompatibleTypes {
         new ValueAndCompatibleTypes(ESTestCase::randomLong, LONG, FLOAT, DOUBLE, BOOLEAN),
         new ValueAndCompatibleTypes(ESTestCase::randomFloat, FLOAT, LONG, DOUBLE, BOOLEAN),
         new ValueAndCompatibleTypes(ESTestCase::randomDouble, DOUBLE, LONG, FLOAT, BOOLEAN),
-        new ValueAndCompatibleTypes(() -> randomAlphaOfLength(5), KEYWORD)
+        new ValueAndCompatibleTypes(() -> BytesRefs.toBytesRef(randomAlphaOfLength(5)), KEYWORD)
     );
 
     public static Literal randomLiteral() {
@@ -129,14 +129,14 @@ public void testReplaceChildren() {
 
     public void testToString() {
         assertThat(new Literal(Source.EMPTY, 1, LONG).toString(), equalTo("1"));
-        assertThat(new Literal(Source.EMPTY, "short", KEYWORD).toString(), equalTo("short"));
+        assertThat(new Literal(Source.EMPTY, BytesRefs.toBytesRef("short"), KEYWORD).toString(), equalTo("short"));
         // toString should limit it's length
         String tooLong = Strings.repeat('a', 510);
-        assertThat(new Literal(Source.EMPTY, tooLong, KEYWORD).toString(), equalTo(Strings.repeat('a', 500) + "..."));
+        assertThat(new Literal(Source.EMPTY, BytesRefs.toBytesRef(tooLong), KEYWORD).toString(), equalTo(Strings.repeat('a', 500) + "..."));
 
         for (ValueAndCompatibleTypes g : GENERATORS) {
             Literal lit = new Literal(Source.EMPTY, g.valueSupplier.get(), randomFrom(g.validDataTypes));
-            assertThat(lit.toString(), equalTo(Objects.toString(lit.value())));
+            assertThat(lit.toString(), equalTo(BytesRefs.toString(lit.value())));
         }
     }
 
diff --git a/x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/expression/function/scalar/FunctionTestUtils.java b/x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/expression/function/scalar/FunctionTestUtils.java
index 9e3ab40ec6462..11befc5c490a0 100644
--- a/x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/expression/function/scalar/FunctionTestUtils.java
+++ b/x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/expression/function/scalar/FunctionTestUtils.java
@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.core.expression.function.scalar;
 
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 
@@ -15,10 +16,13 @@
 public final class FunctionTestUtils {
 
     public static Literal l(Object value) {
-        return new Literal(EMPTY, value, DataType.fromJava(value));
+        return l(value, DataType.fromJava(value));
     }
 
     public static Literal l(Object value, DataType type) {
+        if ((type == DataType.TEXT || type == DataType.KEYWORD) && value instanceof String) {
+            value = BytesRefs.toBytesRef(value);
+        }
         return new Literal(EMPTY, value, type);
     }
 }
diff --git a/x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/util/TestUtils.java b/x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/util/TestUtils.java
index 27b2b9fb620a2..8f07d40bd0665 100644
--- a/x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/util/TestUtils.java
+++ b/x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/util/TestUtils.java
@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.core.util;
 
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
 import org.elasticsearch.xpack.esql.core.tree.Source;
@@ -22,6 +23,8 @@
 import static org.elasticsearch.test.ESTestCase.randomFrom;
 import static org.elasticsearch.xpack.esql.core.tree.Source.EMPTY;
 import static org.elasticsearch.xpack.esql.core.type.DataType.INTEGER;
+import static org.elasticsearch.xpack.esql.core.type.DataType.KEYWORD;
+import static org.elasticsearch.xpack.esql.core.type.DataType.TEXT;
 
 public final class TestUtils {
     private TestUtils() {}
@@ -39,6 +42,10 @@ public static Literal of(Source source, Object value) {
         if (value instanceof Literal) {
             return (Literal) value;
         }
+        DataType type = DataType.fromJava(value);
+        if ((type == TEXT || type == KEYWORD) && value instanceof String) {
+            value = BytesRefs.toBytesRef(value);
+        }
         return new Literal(source, value, DataType.fromJava(value));
     }
 
@@ -58,12 +65,16 @@ public static FieldAttribute getFieldAttribute(String name, DataType dataType) {
         return new FieldAttribute(EMPTY, name, new EsField(name + "f", dataType, emptyMap(), true));
     }
 
-    /** Similar to {@link String#strip()}, but removes the WS throughout the entire string. */
+    /**
+     * Similar to {@link String#strip()}, but removes the WS throughout the entire string.
+     */
     public static String stripThrough(String input) {
         return WS_PATTERN.matcher(input).replaceAll(StringUtils.EMPTY);
     }
 
-    /** Returns the input string, but with parts of it having the letter casing changed. */
+    /**
+     * Returns the input string, but with parts of it having the letter casing changed.
+     */
     public static String randomCasing(String input) {
         StringBuilder sb = new StringBuilder(input.length());
         for (int i = 0, inputLen = input.length(), step = (int) Math.sqrt(inputLen); i < inputLen; i += step) {
diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java b/x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java
index 7b5843704eac8..7ec9ee6344551 100644
--- a/x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java
+++ b/x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java
@@ -21,6 +21,7 @@
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.breaker.NoopCircuitBreaker;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
@@ -227,7 +228,11 @@ public static Literal of(Source source, Object value) {
         if (value instanceof Literal) {
             return (Literal) value;
         }
-        return new Literal(source, value, DataType.fromJava(value));
+        var dataType = DataType.fromJava(value);
+        if (value instanceof String) {
+            value = BytesRefs.toBytesRef(value);
+        }
+        return new Literal(source, value, dataType);
     }
 
     public static ReferenceAttribute referenceAttribute(String name, DataType type) {
diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/ip.csv-spec b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/ip.csv-spec
index 50c46beeff327..86c56909083cb 100644
--- a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/ip.csv-spec
+++ b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/ip.csv-spec
@@ -398,7 +398,7 @@ eth0           |gamma          |fe80::cae2:65ff:fece:feb9
 lo0            |gamma          |fe80::cae2:65ff:fece:feb9                                                         |fe81::cae2:65ff:fece:feb9
 ;
 
-cdirMatchEqualsInsOrsIPs
+cidrMatchEqualsInsOrsIPs
 required_capability: combine_disjunctive_cidrmatches
 
 FROM hosts
diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/match-function.csv-spec b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/match-function.csv-spec
index cd08bb55c0f17..b4b42b7b5a3c7 100644
--- a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/match-function.csv-spec
+++ b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/match-function.csv-spec
@@ -860,3 +860,21 @@ c: long     |   scalerank: long
 10             | 3              
 15             | 2              
 ;
+
+
+testMatchWithReplace
+required_capability: match_function
+required_capability: no_plain_strings_in_literals
+from books 
+| keep book_no, author 
+| where match(author, REPLACE("FaulkneX", "X", "r"))
+| sort book_no 
+| limit 5;
+
+book_no:keyword | author:text
+2378            | [Carol Faulkner, Holly Byers Ochoa, Lucretia Mott]
+2713            | William Faulkner
+2847            | Colleen Faulkner
+2883            | William Faulkner
+3293            | Danny Faulkner
+;
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java
index f685804ae6ad3..aa833016f8612 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java
@@ -1207,7 +1207,15 @@ public enum Cap {
         /**
          * Support parameters for SAMPLE command.
          */
-        PARAMETER_FOR_SAMPLE(Build.current().isSnapshot());
+        PARAMETER_FOR_SAMPLE(Build.current().isSnapshot()),
+
+        /**
+         * From now, Literal only accepts strings as BytesRefs.
+         * No java.lang.String anymore.
+         *
+         * https://github.com/elastic/elasticsearch/issues/129322
+         */
+        NO_PLAIN_STRINGS_IN_LITERALS;
 
         private final boolean enabled;
 
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java
index 7faab1493096a..9f041263d302e 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java
@@ -9,6 +9,7 @@
 
 import org.elasticsearch.common.logging.HeaderWarning;
 import org.elasticsearch.common.logging.LoggerMessageFormat;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.index.IndexMode;
@@ -318,7 +319,7 @@ protected LogicalPlan rule(Enrich plan, AnalyzerContext context) {
                 // the policy does not exist
                 return plan;
             }
-            final String policyName = (String) plan.policyName().fold(FoldContext.small() /* TODO remove me */);
+            final String policyName = BytesRefs.toString(plan.policyName().fold(FoldContext.small() /* TODO remove me */));
             final var resolved = context.enrichResolution().getResolvedPolicy(policyName, plan.mode());
             if (resolved != null) {
                 var policy = new EnrichPolicy(resolved.matchType(), null, List.of(), resolved.matchField(), resolved.enrichFields());
@@ -401,7 +402,7 @@ private static class ResolveInference extends ParameterizedAnalyzerRule<Inferenc
         protected LogicalPlan rule(InferencePlan<?> plan, AnalyzerContext context) {
             assert plan.inferenceId().resolved() && plan.inferenceId().foldable();
 
-            String inferenceId = plan.inferenceId().fold(FoldContext.small()).toString();
+            String inferenceId = BytesRefs.toString(plan.inferenceId().fold(FoldContext.small()));
             ResolvedInference resolvedInference = context.inferenceResolution().getResolvedInference(inferenceId);
 
             if (resolvedInference != null && resolvedInference.taskType() == plan.taskType()) {
@@ -431,7 +432,7 @@ protected LogicalPlan rule(Lookup lookup, AnalyzerContext context) {
             // the parser passes the string wrapped in a literal
             Source source = lookup.source();
             Expression tableNameExpression = lookup.tableName();
-            String tableName = lookup.tableName().toString();
+            String tableName = BytesRefs.toString(tableNameExpression.fold(FoldContext.small() /* TODO remove me */));
             Map<String, Map<String, Column>> tables = context.configuration().tables();
             LocalRelation localRelation = null;
 
@@ -1563,18 +1564,22 @@ private static boolean supportsStringImplicitCasting(DataType type) {
         }
 
         private static UnresolvedAttribute unresolvedAttribute(Expression value, String type, Exception e) {
+            String name = BytesRefs.toString(value.fold(FoldContext.small()) /* TODO remove me */);
             String message = LoggerMessageFormat.format(
+                null,
                 "Cannot convert string [{}] to [{}], error [{}]",
-                value.fold(FoldContext.small() /* TODO remove me */),
+                name,
                 type,
                 (e instanceof ParsingException pe) ? pe.getErrorMessage() : e.getMessage()
             );
-            return new UnresolvedAttribute(value.source(), String.valueOf(value.fold(FoldContext.small() /* TODO remove me */)), message);
+            return new UnresolvedAttribute(value.source(), name, message);
         }
 
         private static Expression castStringLiteralToTemporalAmount(Expression from) {
             try {
-                TemporalAmount result = maybeParseTemporalAmount(from.fold(FoldContext.small() /* TODO remove me */).toString().strip());
+                TemporalAmount result = maybeParseTemporalAmount(
+                    BytesRefs.toString(from.fold(FoldContext.small() /* TODO remove me */)).strip()
+                );
                 if (result == null) {
                     return from;
                 }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/Count.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/Count.java
index af2b975b9e417..aef221ab6a7e9 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/Count.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/Count.java
@@ -163,7 +163,7 @@ public Expression surrogate() {
             return new Mul(
                 s,
                 new Coalesce(s, new MvCount(s, field), List.of(new Literal(s, 0, DataType.INTEGER))),
-                new Count(s, new Literal(s, StringUtils.WILDCARD, DataType.KEYWORD))
+                new Count(s, Literal.keyword(s, StringUtils.WILDCARD))
             );
         }
 
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/Sum.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/Sum.java
index 69795d06ea214..d4a940bbf6648 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/Sum.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/Sum.java
@@ -143,8 +143,6 @@ public Expression surrogate() {
         }
 
         // SUM(const) is equivalent to MV_SUM(const)*COUNT(*).
-        return field.foldable()
-            ? new Mul(s, new MvSum(s, field), new Count(s, new Literal(s, StringUtils.WILDCARD, DataType.KEYWORD)))
-            : null;
+        return field.foldable() ? new Mul(s, new MvSum(s, field), new Count(s, Literal.keyword(s, StringUtils.WILDCARD))) : null;
     }
 }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToIp.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToIp.java
index dd25015a6e6e3..ff8648d86dd38 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToIp.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToIp.java
@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.esql.expression.function.scalar.convert;
 
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.compute.operator.EvalOperator;
 import org.elasticsearch.xpack.esql.core.expression.EntryExpression;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
@@ -39,8 +40,8 @@
 /**
  * Converts strings to IPs.
  * <p>
- *     IPv4 addresses have traditionally parsed quads with leading zeros in three
- *     mutually exclusive ways:
+ * 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>
@@ -189,7 +190,7 @@ protected TypeResolution resolveType() {
                 return new TypeResolution("map keys must be strings");
             }
             if (e.key() instanceof Literal keyl) {
-                key = (String) keyl.value();
+                key = BytesRefs.toString(keyl.value());
             } else {
                 return new TypeResolution("map keys must be literals");
             }
@@ -238,7 +239,7 @@ public static LeadingZeros from(MapExpression exp) {
                 return REJECT;
             }
             Expression e = exp.keyFoldedMap().get(LEADING_ZEROS);
-            return e == null ? REJECT : from((String) ((Literal) e).value());
+            return e == null ? REJECT : from(BytesRefs.toString(((Literal) e).value()));
         }
 
         public static LeadingZeros from(String str) {
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/date/DateDiff.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/date/DateDiff.java
index 19c98f4e5078c..a2ec96d1e0b34 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/date/DateDiff.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/date/DateDiff.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.common.time.DateUtils;
 import org.elasticsearch.compute.ann.Evaluator;
 import org.elasticsearch.compute.ann.Fixed;
@@ -317,7 +318,7 @@ private ExpressionEvaluator.Factory toEvaluator(
 
         if (unit.foldable()) {
             try {
-                Part datePartField = Part.resolve(((BytesRef) unit.fold(toEvaluator.foldCtx())).utf8ToString());
+                Part datePartField = Part.resolve(BytesRefs.toString(unit.fold(toEvaluator.foldCtx())));
                 return constantFactory.build(source(), datePartField, startTimestampEvaluator, endTimestampEvaluator);
             } catch (IllegalArgumentException e) {
                 throw new InvalidArgumentException("invalid unit format for [{}]: {}", sourceText(), e.getMessage());
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvSort.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvSort.java
index 8c272ae70d8d7..4653b5f74dc40 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvSort.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvSort.java
@@ -13,6 +13,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.logging.LoggerMessageFormat;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.BooleanBlock;
@@ -64,8 +65,8 @@ public class MvSort extends EsqlScalarFunction implements OptionalArgument, Post
 
     private final Expression field, order;
 
-    private static final Literal ASC = new Literal(Source.EMPTY, "ASC", DataType.KEYWORD);
-    private static final Literal DESC = new Literal(Source.EMPTY, "DESC", DataType.KEYWORD);
+    private static final Literal ASC = Literal.keyword(Source.EMPTY, "ASC");
+    private static final Literal DESC = Literal.keyword(Source.EMPTY, "DESC");
 
     private static final String INVALID_ORDER_ERROR = "Invalid order value in [{}], expected one of [{}, {}] but got [{}]";
 
@@ -154,14 +155,14 @@ public EvalOperator.ExpressionEvaluator.Factory toEvaluator(ToEvaluator toEvalua
                     null,
                     INVALID_ORDER_ERROR,
                     sourceText(),
-                    ASC.value(),
-                    DESC.value(),
-                    ((BytesRef) order.fold(toEvaluator.foldCtx())).utf8ToString()
+                    BytesRefs.toString(ASC.value()),
+                    BytesRefs.toString(DESC.value()),
+                    BytesRefs.toString(order.fold(toEvaluator.foldCtx()))
                 )
             );
         }
         if (order != null && order.foldable()) {
-            ordering = ((BytesRef) order.fold(toEvaluator.foldCtx())).utf8ToString().equalsIgnoreCase((String) ASC.value());
+            ordering = BytesRefs.toString(order.fold(toEvaluator.foldCtx())).equalsIgnoreCase(BytesRefs.toString(ASC.value()));
         }
 
         return switch (PlannerUtils.toElementType(field.dataType())) {
@@ -243,9 +244,9 @@ public void postOptimizationVerification(Failures failures) {
                     order,
                     INVALID_ORDER_ERROR,
                     sourceText(),
-                    ASC.value(),
-                    DESC.value(),
-                    ((BytesRef) order.fold(FoldContext.small() /* TODO remove me */)).utf8ToString()
+                    BytesRefs.toString(ASC.value()),
+                    BytesRefs.toString(DESC.value()),
+                    BytesRefs.toString(order.fold(FoldContext.small() /* TODO remove me */))
                 )
             );
         }
@@ -261,7 +262,9 @@ private boolean isValidOrder() {
             } else if (obj instanceof String os) {
                 o = os;
             }
-            if (o == null || o.equalsIgnoreCase((String) ASC.value()) == false && o.equalsIgnoreCase((String) DESC.value()) == false) {
+            if (o == null
+                || o.equalsIgnoreCase(BytesRefs.toString(ASC.value())) == false
+                    && o.equalsIgnoreCase(BytesRefs.toString(DESC.value())) == false) {
                 isValidOrder = false;
             }
         }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvZip.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvZip.java
index d6a30c6ca151c..e35f50115b604 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvZip.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvZip.java
@@ -12,6 +12,7 @@
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.compute.ann.Evaluator;
 import org.elasticsearch.compute.data.BytesRefBlock;
 import org.elasticsearch.compute.operator.EvalOperator;
@@ -45,7 +46,7 @@ public class MvZip extends EsqlScalarFunction implements OptionalArgument, Evalu
     public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(Expression.class, "MvZip", MvZip::new);
 
     private final Expression mvLeft, mvRight, delim;
-    private static final Literal COMMA = new Literal(Source.EMPTY, ",", DataType.TEXT);
+    private static final Literal COMMA = new Literal(Source.EMPTY, BytesRefs.toBytesRef(","), DataType.TEXT);
 
     @FunctionInfo(
         returnType = { "keyword" },
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/Replace.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/Replace.java
index 4c7e9fd524263..5da7902d4091d 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/Replace.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/Replace.java
@@ -12,6 +12,7 @@
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.compute.ann.Evaluator;
 import org.elasticsearch.compute.ann.Fixed;
 import org.elasticsearch.compute.operator.EvalOperator.ExpressionEvaluator;
@@ -199,7 +200,7 @@ public ExpressionEvaluator.Factory toEvaluator(ToEvaluator toEvaluator) {
         if (regex.foldable() && regex.dataType() == DataType.KEYWORD) {
             Pattern regexPattern;
             try {
-                regexPattern = Pattern.compile(((BytesRef) regex.fold(toEvaluator.foldCtx())).utf8ToString());
+                regexPattern = Pattern.compile(BytesRefs.toString(regex.fold(toEvaluator.foldCtx())));
             } catch (PatternSyntaxException pse) {
                 // TODO this is not right (inconsistent). See also https://github.com/elastic/elasticsearch/issues/100038
                 // this should generate a header warning and return null (as do the rest of this functionality in evaluators),
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/Space.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/Space.java
index e46c0a730431d..a08ff2860176c 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/Space.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/Space.java
@@ -116,7 +116,7 @@ public ExpressionEvaluator.Factory toEvaluator(ToEvaluator toEvaluator) {
             Object folded = field.fold(toEvaluator.foldCtx());
             if (folded instanceof Integer num) {
                 checkNumber(num);
-                return toEvaluator.apply(new Literal(source(), " ".repeat(num), KEYWORD));
+                return toEvaluator.apply(Literal.keyword(source(), " ".repeat(num)));
             }
         }
 
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/Range.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/Range.java
index 6427449add799..e8953be7348d7 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/Range.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/Range.java
@@ -8,6 +8,7 @@
 
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.logging.LogManager;
 import org.elasticsearch.logging.Logger;
 import org.elasticsearch.xpack.esql.capabilities.TranslationAware;
@@ -166,9 +167,15 @@ class is not serializable (note that writeTo throws UnsupportedOperationExceptio
          */
         if (DataType.isDateTime(value.dataType()) || DataType.isDateTime(lower.dataType()) || DataType.isDateTime(upper.dataType())) {
             try {
+                if (upperValue instanceof BytesRef br) {
+                    upperValue = BytesRefs.toString(br);
+                }
                 if (upperValue instanceof String upperString) {
                     upperValue = asDateTime(upperString);
                 }
+                if (lowerValue instanceof BytesRef br) {
+                    lowerValue = BytesRefs.toString(br);
+                }
                 if (lowerValue instanceof String lowerString) {
                     lowerValue = asDateTime(lowerString);
                 }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/inference/InferenceRunner.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/inference/InferenceRunner.java
index 6d6f52a17d428..d67d6817742c0 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/inference/InferenceRunner.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/inference/InferenceRunner.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.CountDownActionListener;
 import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.inference.TaskType;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.inference.action.GetInferenceModelAction;
@@ -74,7 +75,7 @@ private void resolveInferenceIds(Set<String> inferenceIds, ActionListener<Infere
     }
 
     private static String planInferenceId(InferencePlan<?> plan) {
-        return plan.inferenceId().fold(FoldContext.small()).toString();
+        return BytesRefs.toString(plan.inferenceId().fold(FoldContext.small()));
     }
 
     public void doInference(InferenceAction.Request request, ActionListener<InferenceAction.Response> listener) {
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizer.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizer.java
index 32c764ff29f76..e6665474dfa5d 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizer.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizer.java
@@ -17,7 +17,6 @@
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.CombineEvals;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.CombineProjections;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.ConstantFolding;
-import org.elasticsearch.xpack.esql.optimizer.rules.logical.ConvertStringToByteRef;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.ExtractAggregateCommonFilter;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.FoldNull;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.LiteralsOnTheRight;
@@ -167,7 +166,6 @@ protected static Batch<LogicalPlan> operators() {
             new CombineEvals(),
             new PruneEmptyPlans(),
             new PropagateEmptyRelation(),
-            new ConvertStringToByteRef(),
             new FoldNull(),
             new SplitInWithFoldableValue(),
             new PropagateEvalFoldables(),
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ConvertStringToByteRef.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ConvertStringToByteRef.java
deleted file mode 100644
index b716d8f012d21..0000000000000
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ConvertStringToByteRef.java
+++ /dev/null
@@ -1,47 +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.apache.lucene.util.BytesRef;
-import org.elasticsearch.xpack.esql.core.expression.Expression;
-import org.elasticsearch.xpack.esql.core.expression.Literal;
-import org.elasticsearch.xpack.esql.optimizer.LogicalOptimizerContext;
-
-import java.util.ArrayList;
-import java.util.List;
-
-public final class ConvertStringToByteRef extends OptimizerRules.OptimizerExpressionRule<Literal> {
-
-    public ConvertStringToByteRef() {
-        super(OptimizerRules.TransformDirection.UP);
-    }
-
-    @Override
-    protected Expression rule(Literal lit, LogicalOptimizerContext ctx) {
-        // TODO we shouldn't be emitting String into Literals at all
-        Object value = lit.value();
-
-        if (value == null) {
-            return lit;
-        }
-        if (value instanceof String s) {
-            return Literal.of(lit, new BytesRef(s));
-        }
-        if (value instanceof List<?> l) {
-            if (l.isEmpty() || false == l.get(0) instanceof String) {
-                return lit;
-            }
-            List<BytesRef> byteRefs = new ArrayList<>(l.size());
-            for (Object v : l) {
-                byteRefs.add(new BytesRef(v.toString()));
-            }
-            return Literal.of(lit, byteRefs);
-        }
-        return lit;
-    }
-}
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ReplaceRegexMatch.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ReplaceRegexMatch.java
index 6d92fd40584e7..76bc31510ec3e 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ReplaceRegexMatch.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ReplaceRegexMatch.java
@@ -11,7 +11,6 @@
 import org.elasticsearch.xpack.esql.core.expression.Literal;
 import org.elasticsearch.xpack.esql.core.expression.predicate.regex.RegexMatch;
 import org.elasticsearch.xpack.esql.core.expression.predicate.regex.StringPattern;
-import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.expression.predicate.nulls.IsNotNull;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.Equals;
 import org.elasticsearch.xpack.esql.optimizer.LogicalOptimizerContext;
@@ -43,7 +42,7 @@ public Expression rule(RegexMatch<?> regexMatch, LogicalOptimizerContext ctx) {
         } else {
             String match = pattern.exactMatch();
             if (match != null) {
-                Literal literal = new Literal(regexMatch.source(), match, DataType.KEYWORD);
+                Literal literal = Literal.keyword(regexMatch.source(), match);
                 e = regexToEquals(regexMatch, literal);
             }
         }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/ReplaceFieldWithConstantOrNull.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/ReplaceFieldWithConstantOrNull.java
index 1bc58b9b7e0c3..5635cbba89c31 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/ReplaceFieldWithConstantOrNull.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/ReplaceFieldWithConstantOrNull.java
@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.optimizer.rules.logical.local;
 
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.AttributeSet;
@@ -56,9 +57,9 @@ else if (esRelation.indexMode() == IndexMode.STANDARD) {
                 for (Attribute attribute : esRelation.output()) {
                     if (attribute instanceof FieldAttribute fa) {
                         // Do not use the attribute name, this can deviate from the field name for union types; use fieldName() instead.
-                        var val = localLogicalOptimizerContext.searchStats().constantValue(fa.fieldName());
+                        String val = localLogicalOptimizerContext.searchStats().constantValue(fa.fieldName());
                         if (val != null) {
-                            attrToConstant.put(attribute, Literal.of(attribute, val));
+                            attrToConstant.put(attribute, Literal.of(attribute, BytesRefs.toBytesRef(val)));
                         }
                     }
                 }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/ExpressionBuilder.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/ExpressionBuilder.java
index b32bdbbaab366..cec23786f84dc 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/ExpressionBuilder.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/ExpressionBuilder.java
@@ -16,6 +16,7 @@
 import org.apache.lucene.util.automaton.CharacterRunAutomaton;
 import org.apache.lucene.util.automaton.Operations;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.xpack.esql.core.InvalidArgumentException;
 import org.elasticsearch.xpack.esql.core.expression.Alias;
@@ -81,6 +82,7 @@
 import static org.elasticsearch.xpack.esql.core.type.DataType.DATE_PERIOD;
 import static org.elasticsearch.xpack.esql.core.type.DataType.KEYWORD;
 import static org.elasticsearch.xpack.esql.core.type.DataType.NULL;
+import static org.elasticsearch.xpack.esql.core.type.DataType.TEXT;
 import static org.elasticsearch.xpack.esql.core.type.DataType.TIME_DURATION;
 import static org.elasticsearch.xpack.esql.core.util.NumericUtils.asLongUnsigned;
 import static org.elasticsearch.xpack.esql.core.util.NumericUtils.unsignedLongAsNumber;
@@ -106,15 +108,14 @@ public abstract class ExpressionBuilder extends IdentifierBuilder {
      * eg. EVAL x = sin(sin(sin(sin(sin(sin(sin(sin(sin(....sin(x)....)
      * ANTLR parser is recursive, so the only way to prevent a StackOverflow is to detect how
      * deep we are in the expression parsing and abort the query execution after a threshold
-     *
+     * <p>
      * This value is defined empirically, but the actual stack limit is highly
      * dependent on the JVM and on the JIT.
-     *
+     * <p>
      * A value of 500 proved to be right below the stack limit, but it still triggered
      * some CI failures (once every ~2000 iterations). see https://github.com/elastic/elasticsearch/issues/109846
      * Even though we didn't manage to reproduce the problem in real conditions, we decided
      * to reduce the max allowed depth to 400 (that is still a pretty reasonable limit for real use cases) and be more safe.
-     *
      */
     public static final int MAX_EXPRESSION_DEPTH = 400;
 
@@ -253,7 +254,7 @@ public Literal visitStringLiteral(EsqlBaseParser.StringLiteralContext ctx) {
     @Override
     public Literal visitString(EsqlBaseParser.StringContext ctx) {
         Source source = source(ctx);
-        return new Literal(source, unquote(source), DataType.KEYWORD);
+        return Literal.keyword(source, unquote(source));
     }
 
     @Override
@@ -332,7 +333,7 @@ public NamedExpression visitQualifiedNamePattern(EsqlBaseParser.QualifiedNamePat
                             src,
                             "Query parameter [{}] with value [{}] declared as a constant, cannot be used as an identifier or pattern",
                             ctx.getText(),
-                            lit.value()
+                            lit
                         );
                     }
                 } else if (exp instanceof UnresolvedNamePattern up) {
@@ -622,7 +623,7 @@ public Expression visitFunctionExpression(EsqlBaseParser.FunctionExpressionConte
         if ("count".equals(EsqlFunctionRegistry.normalizeName(name))) {
             // to simplify the registration, handle in the parser the special count cases
             if (args.isEmpty() || ctx.ASTERISK() != null) {
-                args = singletonList(new Literal(source(ctx), "*", DataType.KEYWORD));
+                args = singletonList(Literal.keyword(source(ctx), "*"));
             }
         }
         return new UnresolvedFunction(source(ctx), name, FunctionResolutionStrategy.DEFAULT, args);
@@ -659,7 +660,7 @@ public MapExpression visitMapExpression(EsqlBaseParser.MapExpressionContext ctx)
                 if (l.dataType() == NULL) {
                     throw new ParsingException(source(ctx), "Invalid named function argument [{}], NULL is not supported", entryText);
                 }
-                namedArgs.add(new Literal(source(stringCtx), key, KEYWORD));
+                namedArgs.add(Literal.keyword(source(stringCtx), key));
                 namedArgs.add(l);
                 names.add(key);
             } else {
@@ -744,7 +745,7 @@ public Expression visitRlikeExpression(EsqlBaseParser.RlikeExpressionContext ctx
         Expression left = expression(ctx.valueExpression());
         Literal patternLiteral = visitString(ctx.string());
         try {
-            RLike rLike = new RLike(source, left, new RLikePattern(patternLiteral.fold(FoldContext.small()).toString()));
+            RLike rLike = new RLike(source, left, new RLikePattern(BytesRefs.toString(patternLiteral.fold(FoldContext.small()))));
             return ctx.NOT() == null ? rLike : new Not(source, rLike);
         } catch (InvalidArgumentException e) {
             throw new ParsingException(source, "Invalid pattern for LIKE [{}]: [{}]", patternLiteral, e.getMessage());
@@ -757,7 +758,7 @@ public Expression visitLikeExpression(EsqlBaseParser.LikeExpressionContext ctx)
         Expression left = expression(ctx.valueExpression());
         Literal patternLiteral = visitString(ctx.string());
         try {
-            WildcardPattern pattern = new WildcardPattern(patternLiteral.fold(FoldContext.small()).toString());
+            WildcardPattern pattern = new WildcardPattern(BytesRefs.toString(patternLiteral.fold(FoldContext.small())));
             WildcardLike result = new WildcardLike(source, left, pattern);
             return ctx.NOT() == null ? result : new Not(source, result);
         } catch (InvalidArgumentException e) {
@@ -771,7 +772,7 @@ public Expression visitLikeListExpression(EsqlBaseParser.LikeListExpressionConte
         Expression left = expression(ctx.valueExpression());
         List<WildcardPattern> wildcardPatterns = ctx.string()
             .stream()
-            .map(x -> new WildcardPattern(visitString(x).fold(FoldContext.small()).toString()))
+            .map(x -> new WildcardPattern(BytesRefs.toString(visitString(x).fold(FoldContext.small()))))
             .toList();
         // for now we will use the old WildcardLike function for one argument case to allow compatibility in mixed version deployments
         Expression e = wildcardPatterns.size() == 1
@@ -955,6 +956,9 @@ private Expression visitParam(EsqlBaseParser.ParameterContext ctx, QueryParam pa
                 return new UnresolvedAttribute(source, value.toString());
             }
         }
+        if ((type == KEYWORD || type == TEXT) && value instanceof String) {
+            value = BytesRefs.toBytesRef(value);
+        }
         return new Literal(source, value, type);
     }
 
@@ -1010,7 +1014,7 @@ String unresolvedAttributeNameInParam(ParserRuleContext ctx, Expression param) {
                 source(ctx),
                 invalidParam,
                 ctx.getText(),
-                lit.value() != null ? " with value [" + lit.value() + "] declared as a constant" : " is null or undefined"
+                lit.value() != null ? " with value [" + lit + "] declared as a constant" : " is null or undefined"
             );
             case UnresolvedNamePattern up -> throw new ParsingException(
                 source(ctx),
@@ -1048,9 +1052,9 @@ public Expression visitInputNamedOrPositionalDoubleParams(EsqlBaseParser.InputNa
     }
 
     /**
-      * Double parameter markers represent identifiers, e.g. field or function names. An {@code UnresolvedAttribute}
-      * is returned regardless how the param is specified in the request.
-      */
+     * Double parameter markers represent identifiers, e.g. field or function names. An {@code UnresolvedAttribute}
+     * is returned regardless how the param is specified in the request.
+     */
     private Expression visitDoubleParam(EsqlBaseParser.DoubleParameterContext ctx, QueryParam param) {
         if (param.classification() == PATTERN) {
             context.params.addParsingError(
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/LogicalPlanBuilder.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/LogicalPlanBuilder.java
index c3f0771919e0a..79c566eafa766 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/LogicalPlanBuilder.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/LogicalPlanBuilder.java
@@ -10,9 +10,11 @@
 import org.antlr.v4.runtime.ParserRuleContext;
 import org.antlr.v4.runtime.Token;
 import org.antlr.v4.runtime.tree.ParseTree;
+import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.Build;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.dissect.DissectException;
 import org.elasticsearch.dissect.DissectParser;
@@ -173,7 +175,7 @@ public PlanFactory visitEvalCommand(EsqlBaseParser.EvalCommandContext ctx) {
     public PlanFactory visitGrokCommand(EsqlBaseParser.GrokCommandContext ctx) {
         return p -> {
             Source source = source(ctx);
-            String pattern = visitString(ctx.string()).fold(FoldContext.small() /* TODO remove me */).toString();
+            String pattern = BytesRefs.toString(visitString(ctx.string()).fold(FoldContext.small() /* TODO remove me */));
             Grok.Parser grokParser;
             try {
                 grokParser = Grok.pattern(source, pattern);
@@ -204,21 +206,21 @@ private void validateGrokPattern(Source source, Grok.Parser grokParser, String p
     @Override
     public PlanFactory visitDissectCommand(EsqlBaseParser.DissectCommandContext ctx) {
         return p -> {
-            String pattern = visitString(ctx.string()).fold(FoldContext.small() /* TODO remove me */).toString();
+            String pattern = BytesRefs.toString(visitString(ctx.string()).fold(FoldContext.small() /* TODO remove me */));
             Map<String, Object> options = visitCommandOptions(ctx.commandOptions());
             String appendSeparator = "";
             for (Map.Entry<String, Object> item : options.entrySet()) {
                 if (item.getKey().equalsIgnoreCase("append_separator") == false) {
                     throw new ParsingException(source(ctx), "Invalid option for dissect: [{}]", item.getKey());
                 }
-                if (item.getValue() instanceof String == false) {
+                if (item.getValue() instanceof BytesRef == false) {
                     throw new ParsingException(
                         source(ctx),
                         "Invalid value for dissect append_separator: expected a string, but was [{}]",
                         item.getValue()
                     );
                 }
-                appendSeparator = (String) item.getValue();
+                appendSeparator = BytesRefs.toString(item.getValue());
             }
             Source src = source(ctx);
 
@@ -386,13 +388,17 @@ public PlanFactory visitLimitCommand(EsqlBaseParser.LimitCommandContext ctx) {
         Object val = expression(ctx.constant()).fold(FoldContext.small() /* TODO remove me */);
         if (val instanceof Integer i) {
             if (i < 0) {
-                throw new ParsingException(source, "Invalid value for LIMIT [" + val + "], expecting a non negative integer");
+                throw new ParsingException(source, "Invalid value for LIMIT [" + i + "], expecting a non negative integer");
             }
             return input -> new Limit(source, new Literal(source, i, DataType.INTEGER), input);
         } else {
             throw new ParsingException(
                 source,
-                "Invalid value for LIMIT [" + val + ": " + val.getClass().getSimpleName() + "], expecting a non negative integer"
+                "Invalid value for LIMIT ["
+                    + BytesRefs.toString(val)
+                    + ": "
+                    + (expression(ctx.constant()).dataType() == KEYWORD ? "String" : val.getClass().getSimpleName())
+                    + "], expecting a non negative integer"
             );
         }
     }
@@ -474,7 +480,7 @@ public PlanFactory visitEnrichCommand(EsqlBaseParser.EnrichCommandContext ctx) {
                 source,
                 p,
                 mode,
-                new Literal(source(ctx.policyName), policyNameString, DataType.KEYWORD),
+                Literal.keyword(source(ctx.policyName), policyNameString),
                 matchField,
                 null,
                 Map.of(),
@@ -556,7 +562,7 @@ public PlanFactory visitLookupCommand(EsqlBaseParser.LookupCommandContext ctx) {
             }
         });
 
-        Literal tableName = new Literal(source, visitIndexPattern(List.of(ctx.indexPattern())), DataType.KEYWORD);
+        Literal tableName = Literal.keyword(source, visitIndexPattern(List.of(ctx.indexPattern())));
 
         return p -> new Lookup(source, p, tableName, matchFields, null /* localRelation will be resolved later*/);
     }
@@ -658,7 +664,7 @@ public List<PlanFactory> visitForkSubQueries(EsqlBaseParser.ForkSubQueriesContex
 
         for (var subQueryCtx : ctx.forkSubQuery()) {
             var subQuery = visitForkSubQuery(subQueryCtx);
-            var literal = new Literal(source(ctx), "fork" + count++, KEYWORD);
+            var literal = Literal.keyword(source(ctx), "fork" + count++);
 
             // align _fork id across all fork branches
             Alias alias = null;
@@ -740,9 +746,7 @@ public PlanFactory visitRerankCommand(EsqlBaseParser.RerankCommandContext ctx) {
             );
         }
 
-        Literal inferenceId = ctx.inferenceId != null
-            ? inferenceId(ctx.inferenceId)
-            : new Literal(source, Rerank.DEFAULT_INFERENCE_ID, KEYWORD);
+        Literal inferenceId = ctx.inferenceId != null ? inferenceId(ctx.inferenceId) : Literal.keyword(source, Rerank.DEFAULT_INFERENCE_ID);
 
         return p -> new Rerank(source, p, inferenceId, queryText, visitRerankFields(ctx.rerankFields()));
     }
@@ -761,7 +765,7 @@ public PlanFactory visitCompletionCommand(EsqlBaseParser.CompletionCommandContex
 
     public Literal inferenceId(EsqlBaseParser.IdentifierOrParameterContext ctx) {
         if (ctx.identifier() != null) {
-            return new Literal(source(ctx), visitIdentifier(ctx.identifier()), KEYWORD);
+            return Literal.keyword(source(ctx), visitIdentifier(ctx.identifier()));
         }
 
         if (expression(ctx.parameter()) instanceof Literal literalParam) {
@@ -791,7 +795,7 @@ public PlanFactory visitSampleCommand(EsqlBaseParser.SampleCommandContext ctx) {
         } else {
             throw new ParsingException(
                 source(ctx),
-                "invalid value for SAMPLE probability [" + val + "], expecting a number between 0 and 1, exclusive"
+                "invalid value for SAMPLE probability [" + BytesRefs.toString(val) + "], expecting a number between 0 and 1, exclusive"
             );
         }
     }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/inference/Rerank.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/inference/Rerank.java
index fcfcfb4f21d18..7156c5c97a8fe 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/inference/Rerank.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/inference/Rerank.java
@@ -41,7 +41,7 @@
 public class Rerank extends InferencePlan<Rerank> implements SortAgnostic, SurrogateLogicalPlan, TelemetryAware {
 
     public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(LogicalPlan.class, "Rerank", Rerank::new);
-    public static final Object DEFAULT_INFERENCE_ID = ".rerank-v1-elasticsearch";
+    public static final String DEFAULT_INFERENCE_ID = ".rerank-v1-elasticsearch";
     private final Attribute scoreAttribute;
     private final Expression queryText;
     private final List<Alias> rerankFields;
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java
index 83d071361ecfe..db24eab373463 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java
@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.planner;
 
+import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.common.settings.Settings;
@@ -510,7 +511,8 @@ private PhysicalOperation planTopN(TopNExec topNExec, LocalExecutionPlannerConte
 
         int limit;
         if (topNExec.limit() instanceof Literal literal) {
-            limit = stringToInt(literal.value().toString());
+            Object val = literal.value() instanceof BytesRef br ? BytesRefs.toString(br) : literal.value();
+            limit = stringToInt(val.toString());
         } else {
             throw new EsqlIllegalArgumentException("limit only supported with literal values");
         }
@@ -881,22 +883,30 @@ public static class PhysicalOperation implements Describable {
 
         final Layout layout; // maps field names to channels
 
-        /** Creates a new physical operation with the given source and layout. */
+        /**
+         * Creates a new physical operation with the given source and layout.
+         */
         static PhysicalOperation fromSource(SourceOperatorFactory sourceOperatorFactory, Layout layout) {
             return new PhysicalOperation(sourceOperatorFactory, layout);
         }
 
-        /** Creates a new physical operation from this operation with the given layout. */
+        /**
+         * Creates a new physical operation from this operation with the given layout.
+         */
         PhysicalOperation with(Layout layout) {
             return new PhysicalOperation(this, Optional.empty(), Optional.empty(), layout);
         }
 
-        /** Creates a new physical operation from this operation with the given intermediate operator and layout. */
+        /**
+         * Creates a new physical operation from this operation with the given intermediate operator and layout.
+         */
         PhysicalOperation with(OperatorFactory operatorFactory, Layout layout) {
             return new PhysicalOperation(this, Optional.of(operatorFactory), Optional.empty(), layout);
         }
 
-        /** Creates a new physical operation from this operation with the given sink and layout. */
+        /**
+         * Creates a new physical operation from this operation with the given sink and layout.
+         */
         PhysicalOperation withSink(SinkOperatorFactory sink, Layout layout) {
             return new PhysicalOperation(this, Optional.empty(), Optional.of(sink), layout);
         }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java
index ef9d556cc27e0..7c5e5f8b0a36c 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java
@@ -13,6 +13,7 @@
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.Iterators;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.Page;
@@ -333,7 +334,7 @@ public void analyzedPlan(
         var unresolvedPolicies = preAnalysis.enriches.stream()
             .map(
                 e -> new EnrichPolicyResolver.UnresolvedPolicy(
-                    (String) e.policyName().fold(FoldContext.small() /* TODO remove me*/),
+                    BytesRefs.toString(e.policyName().fold(FoldContext.small() /* TODO remove me*/)),
                     e.mode()
                 )
             )
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/type/EsqlDataTypeConverter.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/type/EsqlDataTypeConverter.java
index 54f796dee0909..18fa0861f5573 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/type/EsqlDataTypeConverter.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/type/EsqlDataTypeConverter.java
@@ -772,16 +772,16 @@ public enum EsqlConverter implements Converter {
         STRING_TO_DATE_PERIOD(x -> EsqlDataTypeConverter.parseTemporalAmount(x, DataType.DATE_PERIOD)),
         STRING_TO_TIME_DURATION(x -> EsqlDataTypeConverter.parseTemporalAmount(x, DataType.TIME_DURATION)),
         STRING_TO_CHRONO_FIELD(EsqlDataTypeConverter::stringToChrono),
-        STRING_TO_DATETIME(x -> EsqlDataTypeConverter.dateTimeToLong((String) x)),
-        STRING_TO_DATE_NANOS(x -> EsqlDataTypeConverter.dateNanosToLong((String) x)),
-        STRING_TO_IP(x -> EsqlDataTypeConverter.stringToIP((String) x)),
-        STRING_TO_VERSION(x -> EsqlDataTypeConverter.stringToVersion((String) x)),
-        STRING_TO_DOUBLE(x -> EsqlDataTypeConverter.stringToDouble((String) x)),
-        STRING_TO_LONG(x -> EsqlDataTypeConverter.stringToLong((String) x)),
-        STRING_TO_INT(x -> EsqlDataTypeConverter.stringToInt((String) x)),
-        STRING_TO_BOOLEAN(x -> EsqlDataTypeConverter.stringToBoolean((String) x)),
-        STRING_TO_GEO(x -> EsqlDataTypeConverter.stringToGeo((String) x)),
-        STRING_TO_SPATIAL(x -> EsqlDataTypeConverter.stringToSpatial((String) x));
+        STRING_TO_DATETIME(x -> EsqlDataTypeConverter.dateTimeToLong(BytesRefs.toString(x))),
+        STRING_TO_DATE_NANOS(x -> EsqlDataTypeConverter.dateNanosToLong(BytesRefs.toString(x))),
+        STRING_TO_IP(x -> EsqlDataTypeConverter.stringToIP(BytesRefs.toString(x))),
+        STRING_TO_VERSION(x -> EsqlDataTypeConverter.stringToVersion(BytesRefs.toString(x))),
+        STRING_TO_DOUBLE(x -> EsqlDataTypeConverter.stringToDouble(BytesRefs.toString(x))),
+        STRING_TO_LONG(x -> EsqlDataTypeConverter.stringToLong(BytesRefs.toString(x))),
+        STRING_TO_INT(x -> EsqlDataTypeConverter.stringToInt(BytesRefs.toString(x))),
+        STRING_TO_BOOLEAN(x -> EsqlDataTypeConverter.stringToBoolean(BytesRefs.toString(x))),
+        STRING_TO_GEO(x -> EsqlDataTypeConverter.stringToGeo(BytesRefs.toString(x))),
+        STRING_TO_SPATIAL(x -> EsqlDataTypeConverter.stringToSpatial(BytesRefs.toString(x)));
 
         private static final String NAME = "esql-converter";
         private final Function<Object, Object> converter;
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/AnalyzerTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/AnalyzerTests.java
index 1c558cbd0a96b..013f1ce201b01 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/AnalyzerTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/AnalyzerTests.java
@@ -12,6 +12,7 @@
 import org.elasticsearch.action.fieldcaps.FieldCapabilitiesResponse;
 import org.elasticsearch.action.fieldcaps.IndexFieldCapabilities;
 import org.elasticsearch.action.fieldcaps.IndexFieldCapabilitiesBuilder;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.analysis.IndexAnalyzers;
@@ -2119,7 +2120,7 @@ public void testLookup() {
         assertThat(as(limit.limit(), Literal.class).value(), equalTo(1000));
 
         var lookup = as(limit.child(), Lookup.class);
-        assertThat(as(lookup.tableName(), Literal.class).value(), equalTo("int_number_names"));
+        assertThat(as(lookup.tableName(), Literal.class).value(), equalTo(BytesRefs.toBytesRef("int_number_names")));
         assertMap(lookup.matchFields().stream().map(Object::toString).toList(), matchesList().item(startsWith("int{r}")));
         assertThat(
             lookup.localRelation().output().stream().map(Object::toString).toList(),
@@ -2870,7 +2871,7 @@ public void testFunctionNamedParamsAsFunctionArgument() {
         MapExpression me = as(match.options(), MapExpression.class);
         assertEquals(1, me.entryExpressions().size());
         EntryExpression ee = as(me.entryExpressions().get(0), EntryExpression.class);
-        assertEquals(new Literal(EMPTY, "minimum_should_match", DataType.KEYWORD), ee.key());
+        assertEquals(new Literal(EMPTY, BytesRefs.toBytesRef("minimum_should_match"), DataType.KEYWORD), ee.key());
         assertEquals(new Literal(EMPTY, 2.0, DataType.DOUBLE), ee.value());
         assertEquals(DataType.DOUBLE, ee.dataType());
     }
@@ -2886,7 +2887,7 @@ public void testFunctionNamedParamsAsFunctionArgument1() {
         MapExpression me = as(qstr.options(), MapExpression.class);
         assertEquals(1, me.entryExpressions().size());
         EntryExpression ee = as(me.entryExpressions().get(0), EntryExpression.class);
-        assertEquals(new Literal(EMPTY, "minimum_should_match", DataType.KEYWORD), ee.key());
+        assertEquals(new Literal(EMPTY, BytesRefs.toBytesRef("minimum_should_match"), DataType.KEYWORD), ee.key());
         assertEquals(new Literal(EMPTY, 3.0, DataType.DOUBLE), ee.value());
         assertEquals(DataType.DOUBLE, ee.dataType());
     }
@@ -2902,7 +2903,7 @@ public void testFunctionNamedParamsAsFunctionArgument2() {
         MapExpression me = as(mm.options(), MapExpression.class);
         assertEquals(1, me.entryExpressions().size());
         EntryExpression ee = as(me.entryExpressions().get(0), EntryExpression.class);
-        assertEquals(new Literal(EMPTY, "minimum_should_match", DataType.KEYWORD), ee.key());
+        assertEquals(new Literal(EMPTY, BytesRefs.toBytesRef("minimum_should_match"), DataType.KEYWORD), ee.key());
         assertEquals(new Literal(EMPTY, 3.0, DataType.DOUBLE), ee.value());
         assertEquals(DataType.DOUBLE, ee.dataType());
     }
@@ -4118,7 +4119,7 @@ static Alias alias(String name, Expression value) {
     }
 
     static Literal string(String value) {
-        return new Literal(EMPTY, value, DataType.KEYWORD);
+        return new Literal(EMPTY, BytesRefs.toBytesRef(value), DataType.KEYWORD);
     }
 
     static Literal literal(int value) {
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/AbstractFunctionTestCase.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/AbstractFunctionTestCase.java
index 750b394fb5044..6c118690ffc6e 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/AbstractFunctionTestCase.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/AbstractFunctionTestCase.java
@@ -12,6 +12,7 @@
 
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.common.breaker.CircuitBreaker;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.MockBigArrays;
@@ -523,7 +524,7 @@ protected Expression serializeDeserializeExpression(Expression expression) {
         // Fields use synthetic sources, which can't be serialized. So we replace with the originals instead.
         var dummyChildren = newExpression.children()
             .stream()
-            .<Expression>map(c -> new Literal(Source.EMPTY, "anything that won't match any test case", c.dataType()))
+            .<Expression>map(c -> new Literal(Source.EMPTY, BytesRefs.toBytesRef("anything that won't match any test case"), c.dataType()))
             .toList();
         // We first replace them with other unrelated expressions to force a replace, as some replaceChildren() will check for equality
         return newExpression.replaceChildrenSameSize(dummyChildren).replaceChildrenSameSize(expression.children());
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/TestCaseSupplier.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/TestCaseSupplier.java
index 66ca512eb135d..584f5f013563e 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/TestCaseSupplier.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/TestCaseSupplier.java
@@ -9,6 +9,7 @@
 
 import org.apache.lucene.document.InetAddressPoint;
 import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.common.time.DateUtils;
 import org.elasticsearch.geo.GeometryTestUtils;
@@ -820,7 +821,7 @@ public static void unary(
     /**
      * Generate cases for {@link DataType#INTEGER}.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#intCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#intCases}.
      * </p>
      */
     public static List<TypedDataSupplier> intCases(int min, int max, boolean includeZero) {
@@ -850,7 +851,7 @@ public static List<TypedDataSupplier> intCases(int min, int max, boolean include
     /**
      * Generate cases for {@link DataType#LONG}.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#longCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#longCases}.
      * </p>
      */
     public static List<TypedDataSupplier> longCases(long min, long max, boolean includeZero) {
@@ -881,7 +882,7 @@ public static List<TypedDataSupplier> longCases(long min, long max, boolean incl
     /**
      * Generate cases for {@link DataType#UNSIGNED_LONG}.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#ulongCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#ulongCases}.
      * </p>
      */
     public static List<TypedDataSupplier> ulongCases(BigInteger min, BigInteger max, boolean includeZero) {
@@ -927,7 +928,7 @@ public static List<TypedDataSupplier> ulongCases(BigInteger min, BigInteger max,
     /**
      * Generate cases for {@link DataType#DOUBLE}.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#doubleCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#doubleCases}.
      * </p>
      */
     public static List<TypedDataSupplier> doubleCases(double min, double max, boolean includeZero) {
@@ -997,7 +998,7 @@ public static List<TypedDataSupplier> doubleCases(double min, double max, boolea
     /**
      * Generate cases for {@link DataType#BOOLEAN}.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#booleanCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#booleanCases}.
      * </p>
      */
     public static List<TypedDataSupplier> booleanCases() {
@@ -1010,7 +1011,7 @@ public static List<TypedDataSupplier> booleanCases() {
     /**
      * Generate cases for {@link DataType#DATETIME}.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#dateCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#dateCases}.
      * </p>
      */
     public static List<TypedDataSupplier> dateCases() {
@@ -1020,7 +1021,7 @@ public static List<TypedDataSupplier> dateCases() {
     /**
      * Generate cases for {@link DataType#DATETIME}.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#dateCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#dateCases}.
      * </p>
      * Helper function for if you want to specify your min and max range as dates instead of longs.
      */
@@ -1031,7 +1032,7 @@ public static List<TypedDataSupplier> dateCases(Instant min, Instant max) {
     /**
      * Generate cases for {@link DataType#DATETIME}.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#dateCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#dateCases}.
      * </p>
      */
     public static List<TypedDataSupplier> dateCases(long min, long max) {
@@ -1067,7 +1068,6 @@ public static List<TypedDataSupplier> dateCases(long min, long max) {
     }
 
     /**
-     *
      * @return randomized valid date formats
      */
     public static List<TypedDataSupplier> dateFormatCases() {
@@ -1081,7 +1081,6 @@ public static List<TypedDataSupplier> dateFormatCases() {
 
     /**
      * Generate cases for {@link DataType#DATE_NANOS}.
-     *
      */
     public static List<TypedDataSupplier> dateNanosCases() {
         return dateNanosCases(Instant.EPOCH, DateUtils.MAX_NANOSECOND_INSTANT);
@@ -1089,7 +1088,6 @@ public static List<TypedDataSupplier> dateNanosCases() {
 
     /**
      * Generate cases for {@link DataType#DATE_NANOS}.
-     *
      */
     public static List<TypedDataSupplier> dateNanosCases(Instant minValue, Instant maxValue) {
         // maximum nanosecond date in ES is 2262-04-11T23:47:16.854775807Z
@@ -1206,7 +1204,7 @@ public static List<TypedDataSupplier> cartesianShapeCases() {
     /**
      * Generate cases for {@link DataType#GEO_POINT}.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#geoPointCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#geoPointCases}.
      * </p>
      */
     public static List<TypedDataSupplier> geoPointCases(Supplier<Boolean> hasAlt) {
@@ -1218,7 +1216,7 @@ public static List<TypedDataSupplier> geoPointCases(Supplier<Boolean> hasAlt) {
     /**
      * Generate cases for {@link DataType#CARTESIAN_POINT}.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#cartesianPointCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#cartesianPointCases}.
      * </p>
      */
     public static List<TypedDataSupplier> cartesianPointCases(Supplier<Boolean> hasAlt) {
@@ -1254,7 +1252,7 @@ public static List<TypedDataSupplier> cartesianShapeCases(Supplier<Boolean> hasA
     /**
      * Generate cases for {@link DataType#IP}.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#ipCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#ipCases}.
      * </p>
      */
     public static List<TypedDataSupplier> ipCases() {
@@ -1272,7 +1270,7 @@ public static List<TypedDataSupplier> ipCases() {
     /**
      * Generate cases for String DataTypes.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#stringCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#stringCases}.
      * </p>
      */
     public static List<TypedDataSupplier> stringCases(DataType type) {
@@ -1304,7 +1302,7 @@ public static List<TypedDataSupplier> stringCases(DataType type) {
     /**
      * Supplier test case data for {@link Version} fields.
      * <p>
-     *     For multi-row parameters, see {@link MultiRowTestCaseSupplier#versionCases}.
+     * For multi-row parameters, see {@link MultiRowTestCaseSupplier#versionCases}.
      * </p>
      */
     public static List<TypedDataSupplier> versionCases(String prefix) {
@@ -1456,6 +1454,7 @@ public TestCase(List<TypedData> data, Matcher<String> evaluatorToString, DataTyp
 
         /**
          * Build a test case for type errors.
+         *
          * @deprecated use a subclass of {@link ErrorsForCasesWithoutExamplesTestCase} instead
          */
         @Deprecated
@@ -1702,7 +1701,7 @@ public TestCase withFoldingException(Class<? extends Throwable> clazz, String me
         /**
          * Build a new {@link TestCase} that can't build an evaluator.
          * <p>
-         *     Useful for special cases that can't be executed, but should still be considered.
+         * Useful for special cases that can't be executed, but should still be considered.
          * </p>
          */
         public TestCase withoutEvaluator() {
@@ -1765,11 +1764,11 @@ public static class TypedData {
         private final boolean mapExpression;
 
         /**
-         * @param data value to test against
-         * @param type type of the value, for building expressions
-         * @param name a name for the value, used for generating test case names
+         * @param data         value to test against
+         * @param type         type of the value, for building expressions
+         * @param name         a name for the value, used for generating test case names
          * @param forceLiteral should this data always be converted to a literal and <strong>never</strong> to a field reference?
-         * @param multiRow if true, data is expected to be a List of values, one per row
+         * @param multiRow     if true, data is expected to be a List of values, one per row
          */
         private TypedData(Object data, DataType type, String name, boolean forceLiteral, boolean multiRow) {
             assert multiRow == false || data instanceof List : "multiRow data must be a List";
@@ -1798,6 +1797,7 @@ public TypedData(Object data, DataType type, String name) {
 
         /**
          * Build a value, guessing the type via reflection.
+         *
          * @param data value to test against
          * @param name a name for the value, used for generating test case names
          */
@@ -1807,6 +1807,7 @@ public TypedData(Object data, String name) {
 
         /**
          * Create a TypedData object for field to be aggregated.
+         *
          * @param data values to test against, one per row
          * @param type type of the value, for building expressions
          * @param name a name for the value, used for generating test case names
@@ -1886,9 +1887,16 @@ public Literal asLiteral() {
                     throw new IllegalStateException("Multirow values require exactly 1 element to be a literal, got " + values.size());
                 }
 
-                return new Literal(Source.synthetic(name), values.get(0), type);
+                return new Literal(Source.synthetic(name), stringToBytesRef(values.get(0), type), type);
+            }
+            return new Literal(Source.synthetic(name), stringToBytesRef(data, type), type);
+        }
+
+        private Object stringToBytesRef(Object o, DataType type) {
+            if ((type == DataType.KEYWORD || type == DataType.TEXT) && o instanceof String s) {
+                return BytesRefs.toBytesRef(s);
             }
-            return new Literal(Source.synthetic(name), data, type);
+            return o;
         }
 
         /**
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/KnnTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/KnnTests.java
index c2bc381e2663c..76db793c4e772 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/KnnTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/KnnTests.java
@@ -32,7 +32,6 @@
 import static org.elasticsearch.xpack.esql.SerializationTestUtils.serializeDeserialize;
 import static org.elasticsearch.xpack.esql.core.type.DataType.BOOLEAN;
 import static org.elasticsearch.xpack.esql.core.type.DataType.DENSE_VECTOR;
-import static org.elasticsearch.xpack.esql.core.type.DataType.KEYWORD;
 import static org.elasticsearch.xpack.esql.core.type.DataType.UNSUPPORTED;
 import static org.elasticsearch.xpack.esql.planner.TranslatorHandler.TRANSLATOR_HANDLER;
 import static org.hamcrest.Matchers.equalTo;
@@ -91,7 +90,7 @@ private static List<TestCaseSupplier> addFunctionNamedParams(List<TestCaseSuppli
                 List<TestCaseSupplier.TypedData> values = new ArrayList<>(supplier.get().getData());
                 values.add(
                     new TestCaseSupplier.TypedData(
-                        new MapExpression(Source.EMPTY, List.of(new Literal(Source.EMPTY, randomAlphaOfLength(10), KEYWORD))),
+                        new MapExpression(Source.EMPTY, List.of(Literal.keyword(Source.EMPTY, randomAlphaOfLength(10)))),
                         UNSUPPORTED,
                         "options"
                     ).forceLiteral()
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/MatchPhraseTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/MatchPhraseTests.java
index ce996962398bf..6b04ed2f93e88 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/MatchPhraseTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/MatchPhraseTests.java
@@ -30,7 +30,6 @@
 import static org.elasticsearch.xpack.esql.SerializationTestUtils.serializeDeserialize;
 import static org.elasticsearch.xpack.esql.core.type.DataType.BOOLEAN;
 import static org.elasticsearch.xpack.esql.core.type.DataType.INTEGER;
-import static org.elasticsearch.xpack.esql.core.type.DataType.KEYWORD;
 import static org.elasticsearch.xpack.esql.core.type.DataType.UNSUPPORTED;
 import static org.elasticsearch.xpack.esql.expression.function.TestCaseSupplier.stringCases;
 import static org.elasticsearch.xpack.esql.planner.TranslatorHandler.TRANSLATOR_HANDLER;
@@ -87,7 +86,7 @@ private static List<TestCaseSupplier> addFunctionNamedParams(List<TestCaseSuppli
                     new TestCaseSupplier.TypedData(
                         new MapExpression(
                             Source.EMPTY,
-                            List.of(new Literal(Source.EMPTY, "slop", INTEGER), new Literal(Source.EMPTY, randomAlphaOfLength(10), KEYWORD))
+                            List.of(new Literal(Source.EMPTY, "slop", INTEGER), Literal.keyword(Source.EMPTY, randomAlphaOfLength(10)))
                         ),
                         UNSUPPORTED,
                         "options"
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/MatchTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/MatchTests.java
index 301cbd6844afe..cb8833f00a694 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/MatchTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/MatchTests.java
@@ -28,7 +28,6 @@
 
 import static org.elasticsearch.xpack.esql.SerializationTestUtils.serializeDeserialize;
 import static org.elasticsearch.xpack.esql.core.type.DataType.BOOLEAN;
-import static org.elasticsearch.xpack.esql.core.type.DataType.KEYWORD;
 import static org.elasticsearch.xpack.esql.core.type.DataType.UNSUPPORTED;
 import static org.elasticsearch.xpack.esql.planner.TranslatorHandler.TRANSLATOR_HANDLER;
 import static org.hamcrest.Matchers.equalTo;
@@ -59,10 +58,7 @@ private static List<TestCaseSupplier> addFunctionNamedParams(List<TestCaseSuppli
                     new TestCaseSupplier.TypedData(
                         new MapExpression(
                             Source.EMPTY,
-                            List.of(
-                                new Literal(Source.EMPTY, "fuzziness", KEYWORD),
-                                new Literal(Source.EMPTY, randomAlphaOfLength(10), KEYWORD)
-                            )
+                            List.of(Literal.keyword(Source.EMPTY, "fuzziness"), Literal.keyword(Source.EMPTY, randomAlphaOfLength(10)))
                         ),
                         UNSUPPORTED,
                         "options"
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/QueryStringTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/QueryStringTests.java
index 71c0debc13fcc..df8efcfc87e03 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/QueryStringTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/fulltext/QueryStringTests.java
@@ -27,7 +27,6 @@
 
 import static org.elasticsearch.xpack.esql.SerializationTestUtils.serializeDeserialize;
 import static org.elasticsearch.xpack.esql.core.type.DataType.BOOLEAN;
-import static org.elasticsearch.xpack.esql.core.type.DataType.KEYWORD;
 import static org.elasticsearch.xpack.esql.core.type.DataType.UNSUPPORTED;
 import static org.elasticsearch.xpack.esql.planner.TranslatorHandler.TRANSLATOR_HANDLER;
 import static org.hamcrest.Matchers.equalTo;
@@ -58,10 +57,7 @@ private static List<TestCaseSupplier> addFunctionNamedParams(List<TestCaseSuppli
                     new TestCaseSupplier.TypedData(
                         new MapExpression(
                             Source.EMPTY,
-                            List.of(
-                                new Literal(Source.EMPTY, "default_field", KEYWORD),
-                                new Literal(Source.EMPTY, randomAlphaOfLength(10), KEYWORD)
-                            )
+                            List.of(Literal.keyword(Source.EMPTY, "default_field"), Literal.keyword(Source.EMPTY, randomAlphaOfLength(10)))
                         ),
                         UNSUPPORTED,
                         "options"
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/conditional/CaseExtraTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/conditional/CaseExtraTests.java
index 911878a645b42..fd54408e0e2b8 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/conditional/CaseExtraTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/conditional/CaseExtraTests.java
@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.esql.expression.function.scalar.conditional;
 
 import org.elasticsearch.common.breaker.CircuitBreaker;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.MockBigArrays;
@@ -305,7 +306,11 @@ private static Case caseExpr(Object... args) {
             if (arg instanceof Expression e) {
                 return e;
             }
-            return new Literal(Source.synthetic(arg == null ? "null" : arg.toString()), arg, DataType.fromJava(arg));
+            DataType dataType = DataType.fromJava(arg);
+            if (arg instanceof String) {
+                arg = BytesRefs.toBytesRef(arg);
+            }
+            return new Literal(Source.synthetic(arg == null ? "null" : BytesRefs.toString(arg)), arg, dataType);
         }).toList();
         return new Case(Source.synthetic("<case>"), exps.get(0), exps.subList(1, exps.size()));
     }
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToIpTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToIpTests.java
index b41177f83c58d..56f6206d51650 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToIpTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToIpTests.java
@@ -143,8 +143,8 @@ private MapExpression options() {
         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)
+                Literal.keyword(Source.EMPTY, "leading_zeros"),
+                Literal.keyword(Source.EMPTY, leadingZeros.toString().toLowerCase(Locale.ROOT))
             )
         );
     }
@@ -182,8 +182,10 @@ private static String readEvaluator() {
     }
 
     private static String stringEvaluator(ToIp.LeadingZeros leadingZeros) {
+        if (leadingZeros == null) {
+            return "ParseIpLeadingZerosRejectedEvaluator[string=" + readEvaluator() + "]";
+        }
         return switch (leadingZeros) {
-            case null -> "ParseIpLeadingZerosRejectedEvaluator";
             case REJECT -> "ParseIpLeadingZerosRejectedEvaluator";
             case DECIMAL -> "ParseIpLeadingZerosAreDecimalEvaluator";
             case OCTAL -> "ParseIpLeadingZerosAreOctalEvaluator";
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/EndsWithTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/EndsWithTests.java
index d0a68f4986efd..d6a6106f9372e 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/EndsWithTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/EndsWithTests.java
@@ -108,11 +108,7 @@ protected Expression build(Source source, List<Expression> args) {
     }
 
     public void testLuceneQuery_AllLiterals_NonTranslatable() {
-        var function = new EndsWith(
-            Source.EMPTY,
-            new Literal(Source.EMPTY, "test", DataType.KEYWORD),
-            new Literal(Source.EMPTY, "test", DataType.KEYWORD)
-        );
+        var function = new EndsWith(Source.EMPTY, Literal.keyword(Source.EMPTY, "test"), Literal.keyword(Source.EMPTY, "test"));
 
         assertThat(function.translatable(LucenePushdownPredicates.DEFAULT), equalTo(TranslationAware.Translatable.NO));
     }
@@ -131,7 +127,7 @@ public void testLuceneQuery_NonFoldableSuffix_Translatable() {
         var function = new EndsWith(
             Source.EMPTY,
             new FieldAttribute(Source.EMPTY, "field", new EsField("suffix", DataType.KEYWORD, Map.of(), true)),
-            new Literal(Source.EMPTY, "a*b?c\\", DataType.KEYWORD)
+            Literal.keyword(Source.EMPTY, "a*b?c\\")
         );
 
         assertThat(function.translatable(LucenePushdownPredicates.DEFAULT), equalTo(TranslationAware.Translatable.YES));
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/StartsWithTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/StartsWithTests.java
index bbbb717f03eb6..c716457dd8378 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/StartsWithTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/StartsWithTests.java
@@ -68,11 +68,7 @@ protected Expression build(Source source, List<Expression> args) {
     }
 
     public void testLuceneQuery_AllLiterals_NonTranslatable() {
-        var function = new StartsWith(
-            Source.EMPTY,
-            new Literal(Source.EMPTY, "test", DataType.KEYWORD),
-            new Literal(Source.EMPTY, "test", DataType.KEYWORD)
-        );
+        var function = new StartsWith(Source.EMPTY, Literal.keyword(Source.EMPTY, "test"), Literal.keyword(Source.EMPTY, "test"));
 
         assertThat(function.translatable(LucenePushdownPredicates.DEFAULT), equalTo(TranslationAware.Translatable.NO));
     }
@@ -91,7 +87,7 @@ public void testLuceneQuery_NonFoldablePrefix_Translatable() {
         var function = new StartsWith(
             Source.EMPTY,
             new FieldAttribute(Source.EMPTY, "field", new EsField("prefix", DataType.KEYWORD, Map.of(), true)),
-            new Literal(Source.EMPTY, "a*b?c\\", DataType.KEYWORD)
+            Literal.keyword(Source.EMPTY, "a*b?c\\")
         );
 
         assertThat(function.translatable(LucenePushdownPredicates.DEFAULT), equalTo(TranslationAware.Translatable.YES));
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ToLowerTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ToLowerTests.java
index 4b2e3e9774a5b..78d29ab71684d 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ToLowerTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ToLowerTests.java
@@ -54,7 +54,7 @@ public static Iterable<Object[]> parameters() {
     public void testRandomLocale() {
         String testString = randomAlphaOfLength(10);
         Configuration cfg = randomLocaleConfig();
-        ToLower func = new ToLower(Source.EMPTY, new Literal(Source.EMPTY, testString, DataType.KEYWORD), cfg);
+        ToLower func = new ToLower(Source.EMPTY, Literal.keyword(Source.EMPTY, testString), cfg);
         assertThat(BytesRefs.toBytesRef(testString.toLowerCase(cfg.locale())), equalTo(func.fold(FoldContext.small())));
     }
 
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ToUpperTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ToUpperTests.java
index 43c0e393762f5..4ee5147c358e4 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ToUpperTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ToUpperTests.java
@@ -54,7 +54,7 @@ public static Iterable<Object[]> parameters() {
     public void testRandomLocale() {
         String testString = randomAlphaOfLength(10);
         Configuration cfg = randomLocaleConfig();
-        ToUpper func = new ToUpper(Source.EMPTY, new Literal(Source.EMPTY, testString, DataType.KEYWORD), cfg);
+        ToUpper func = new ToUpper(Source.EMPTY, Literal.keyword(Source.EMPTY, testString), cfg);
         assertThat(BytesRefs.toBytesRef(testString.toUpperCase(cfg.locale())), equalTo(func.fold(FoldContext.small())));
     }
 
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/inference/InferenceRunnerTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/inference/InferenceRunnerTests.java
index 1a22da2701b53..ef7b3984bd532 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/inference/InferenceRunnerTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/inference/InferenceRunnerTests.java
@@ -24,7 +24,6 @@
 import org.elasticsearch.xpack.core.inference.action.GetInferenceModelAction;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
 import org.elasticsearch.xpack.esql.core.tree.Source;
-import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.plan.logical.inference.InferencePlan;
 import org.elasticsearch.xpack.esql.plugin.EsqlPlugin;
 import org.junit.After;
@@ -182,7 +181,7 @@ private static ModelConfigurations mockModelConfig(String inferenceId, TaskType
 
     private static InferencePlan<?> mockInferencePlan(String inferenceId) {
         InferencePlan<?> plan = mock(InferencePlan.class);
-        when(plan.inferenceId()).thenReturn(new Literal(Source.EMPTY, inferenceId, DataType.KEYWORD));
+        when(plan.inferenceId()).thenReturn(Literal.keyword(Source.EMPTY, inferenceId));
         return plan;
     }
 }
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java
index 6d631283aee05..e432aabc2bac4 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java
@@ -5111,7 +5111,7 @@ public void testCountOfLiteral() {
 
         assertThat(Expressions.names(agg.aggregates()), contains("$$COUNT$s$0", "w"));
         var countAggLiteral = as(as(Alias.unwrap(agg.aggregates().get(0)), Count.class).field(), Literal.class);
-        assertTrue(countAggLiteral.semanticEquals(new Literal(EMPTY, StringUtils.WILDCARD, DataType.KEYWORD)));
+        assertTrue(countAggLiteral.semanticEquals(new Literal(EMPTY, BytesRefs.toBytesRef(StringUtils.WILDCARD), DataType.KEYWORD)));
 
         var exprs = eval.fields();
         // s == mv_count([1,2]) * count(*)
@@ -5635,7 +5635,7 @@ record PushdownShadowingGeneratingPlanTestCase(
                 EMPTY,
                 plan,
                 Enrich.Mode.ANY,
-                new Literal(EMPTY, "some_policy", KEYWORD),
+                Literal.keyword(EMPTY, "some_policy"),
                 attr,
                 null,
                 Map.of(),
@@ -5674,8 +5674,8 @@ record PushdownShadowingGeneratingPlanTestCase(
      * And similarly for dissect, grok and enrich.
      */
     public void testPushShadowingGeneratingPlanPastProject() {
-        Alias x = new Alias(EMPTY, "x", new Literal(EMPTY, "1", KEYWORD));
-        Alias y = new Alias(EMPTY, "y", new Literal(EMPTY, "2", KEYWORD));
+        Alias x = new Alias(EMPTY, "x", Literal.keyword(EMPTY, "1"));
+        Alias y = new Alias(EMPTY, "y", Literal.keyword(EMPTY, "2"));
         LogicalPlan initialRow = new Row(EMPTY, List.of(x, y));
         LogicalPlan initialProject = new Project(EMPTY, initialRow, List.of(y.toAttribute(), x.toAttribute()));
 
@@ -5721,8 +5721,8 @@ public void testPushShadowingGeneratingPlanPastProject() {
      * And similarly for dissect, grok and enrich.
      */
     public void testPushShadowingGeneratingPlanPastRenamingProject() {
-        Alias x = new Alias(EMPTY, "x", new Literal(EMPTY, "1", KEYWORD));
-        Alias y = new Alias(EMPTY, "y", new Literal(EMPTY, "2", KEYWORD));
+        Alias x = new Alias(EMPTY, "x", Literal.keyword(EMPTY, "1"));
+        Alias y = new Alias(EMPTY, "y", Literal.keyword(EMPTY, "2"));
         LogicalPlan initialRow = new Row(EMPTY, List.of(x, y));
         LogicalPlan initialProject = new Project(
             EMPTY,
@@ -5779,7 +5779,7 @@ public void testPushShadowingGeneratingPlanPastRenamingProject() {
      * And similarly for dissect, grok and enrich.
      */
     public void testPushShadowingGeneratingPlanPastRenamingProjectWithResolution() {
-        Alias y = new Alias(EMPTY, "y", new Literal(EMPTY, "2", KEYWORD));
+        Alias y = new Alias(EMPTY, "y", Literal.keyword(EMPTY, "2"));
         Alias yAliased = new Alias(EMPTY, "x", y.toAttribute());
         LogicalPlan initialRow = new Row(EMPTY, List.of(y));
         LogicalPlan initialProject = new Project(EMPTY, initialRow, List.of(y.toAttribute(), yAliased));
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PushDownAndCombineLimitsTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PushDownAndCombineLimitsTests.java
index f5d88618b352d..6e6ac5e0fa0a8 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PushDownAndCombineLimitsTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PushDownAndCombineLimitsTests.java
@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.optimizer.rules.logical;
 
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.esql.core.expression.Alias;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
@@ -87,7 +88,7 @@ public void checkOptimizedPlan(LogicalPlan basePlan, LogicalPlan optimizedPlan)
     private static final List<PushDownLimitTestCase<? extends UnaryPlan>> NON_PUSHABLE_LIMIT_TEST_CASES = List.of(
         new PushDownLimitTestCase<>(
             Filter.class,
-            (plan, attr) -> new Filter(EMPTY, plan, new Equals(EMPTY, attr, new Literal(EMPTY, "right", TEXT))),
+            (plan, attr) -> new Filter(EMPTY, plan, new Equals(EMPTY, attr, new Literal(EMPTY, BytesRefs.toBytesRef("right"), TEXT))),
             (basePlan, optimizedPlan) -> {
                 assertEquals(basePlan.source(), optimizedPlan.source());
                 assertEquals(basePlan.condition(), optimizedPlan.condition());
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ReplaceRegexMatchTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ReplaceRegexMatchTests.java
index 39926c63e7210..c89611f304671 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ReplaceRegexMatchTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ReplaceRegexMatchTests.java
@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.optimizer.rules.logical;
 
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
@@ -61,7 +62,7 @@ public void testExactMatchWildcardLike() {
             assertEquals(Equals.class, e.getClass());
             Equals eq = (Equals) e;
             assertEquals(fa, eq.left());
-            assertEquals(s.replace("\\", StringUtils.EMPTY), eq.right().fold(FoldContext.small()));
+            assertEquals(s.replace("\\", StringUtils.EMPTY), BytesRefs.toString(eq.right().fold(FoldContext.small())));
         }
     }
 
@@ -73,7 +74,7 @@ public void testExactMatchRLike() {
         assertEquals(Equals.class, e.getClass());
         Equals eq = (Equals) e;
         assertEquals(fa, eq.left());
-        assertEquals("abc", eq.right().fold(FoldContext.small()));
+        assertEquals("abc", BytesRefs.toString(eq.right().fold(FoldContext.small())));
     }
 
 }
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/AbstractStatementParserTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/AbstractStatementParserTests.java
index 27c2b2b67e984..462b9f7373ecf 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/AbstractStatementParserTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/AbstractStatementParserTests.java
@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.esql.parser;
 
 import org.elasticsearch.common.logging.LoggerMessageFormat;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.esql.VerificationException;
@@ -120,11 +121,11 @@ static Literal literalBooleans(boolean... booleans) {
     }
 
     static Literal literalString(String s) {
-        return new Literal(EMPTY, s, DataType.KEYWORD);
+        return Literal.keyword(EMPTY, s);
     }
 
     static Literal literalStrings(String... strings) {
-        return new Literal(EMPTY, Arrays.asList(strings), DataType.KEYWORD);
+        return new Literal(EMPTY, Arrays.asList(strings).stream().map(BytesRefs::toBytesRef).toList(), DataType.KEYWORD);
     }
 
     static MapExpression mapExpression(Map<String, Object> keyValuePairs) {
@@ -133,12 +134,24 @@ static MapExpression mapExpression(Map<String, Object> keyValuePairs) {
             String key = entry.getKey();
             Object value = entry.getValue();
             DataType type = (value instanceof List<?> l) ? DataType.fromJava(l.get(0)) : DataType.fromJava(value);
-            ees.add(new Literal(EMPTY, key, DataType.KEYWORD));
+            value = stringsToBytesRef(value, type);
+
+            ees.add(Literal.keyword(EMPTY, key));
             ees.add(new Literal(EMPTY, value, type));
         }
         return new MapExpression(EMPTY, ees);
     }
 
+    private static Object stringsToBytesRef(Object value, DataType type) {
+        if (value instanceof List<?> l) {
+            return l.stream().map(x -> stringsToBytesRef(x, type)).toList();
+        }
+        if (value instanceof String && (type == DataType.TEXT || type == DataType.KEYWORD)) {
+            value = BytesRefs.toBytesRef(value);
+        }
+        return value;
+    }
+
     void expectError(String query, String errorMessage) {
         expectError(query, null, errorMessage);
     }
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/ExpressionTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/ExpressionTests.java
index aebd847aec44e..c39dc70f3ad39 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/ExpressionTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/ExpressionTests.java
@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.parser;
 
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.esql.core.expression.Alias;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
@@ -49,6 +50,7 @@
 import static org.elasticsearch.xpack.esql.core.type.DataType.INTEGER;
 import static org.elasticsearch.xpack.esql.core.type.DataType.KEYWORD;
 import static org.elasticsearch.xpack.esql.core.type.DataType.LONG;
+import static org.elasticsearch.xpack.esql.core.type.DataType.TEXT;
 import static org.elasticsearch.xpack.esql.core.type.DataType.TIME_DURATION;
 import static org.elasticsearch.xpack.esql.expression.function.FunctionResolutionStrategy.DEFAULT;
 import static org.hamcrest.Matchers.containsString;
@@ -661,6 +663,9 @@ private LogicalPlan parse(String s) {
     }
 
     private Literal l(Object value, DataType type) {
+        if (value instanceof String && (type == TEXT || type == KEYWORD)) {
+            value = BytesRefs.toBytesRef(value);
+        }
         return new Literal(null, value, type);
     }
 
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/StatementParserTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/StatementParserTests.java
index d8709de54c346..c8056fb2601a6 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/StatementParserTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/StatementParserTests.java
@@ -7,8 +7,10 @@
 
 package org.elasticsearch.xpack.esql.parser;
 
+import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.Build;
 import org.elasticsearch.common.logging.LoggerMessageFormat;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.xpack.esql.action.EsqlCapabilities;
@@ -223,7 +225,7 @@ public void testRowCommandWithEscapedFieldName() {
                 List.of(
                     new Alias(EMPTY, "a.b.c", integer(1)),
                     new Alias(EMPTY, "b", integer(2)),
-                    new Alias(EMPTY, "@timestamp", new Literal(EMPTY, "2022-26-08T00:00:00", KEYWORD))
+                    new Alias(EMPTY, "@timestamp", Literal.keyword(EMPTY, "2022-26-08T00:00:00"))
                 )
             ),
             statement("row a.b.c = 1, `b` = 2, `@timestamp`=\"2022-26-08T00:00:00\"")
@@ -1225,7 +1227,7 @@ public void testEnrich() {
                 EMPTY,
                 PROCESSING_CMD_INPUT,
                 null,
-                new Literal(EMPTY, "countries", KEYWORD),
+                Literal.keyword(EMPTY, "countries"),
                 new EmptyAttribute(EMPTY),
                 null,
                 Map.of(),
@@ -1239,7 +1241,7 @@ public void testEnrich() {
                 EMPTY,
                 PROCESSING_CMD_INPUT,
                 null,
-                new Literal(EMPTY, "index-policy", KEYWORD),
+                Literal.keyword(EMPTY, "index-policy"),
                 new UnresolvedAttribute(EMPTY, "field_underscore"),
                 null,
                 Map.of(),
@@ -1254,7 +1256,7 @@ public void testEnrich() {
                 EMPTY,
                 PROCESSING_CMD_INPUT,
                 mode,
-                new Literal(EMPTY, "countries", KEYWORD),
+                Literal.keyword(EMPTY, "countries"),
                 new UnresolvedAttribute(EMPTY, "country_code"),
                 null,
                 Map.of(),
@@ -1345,33 +1347,33 @@ public void testInputParams() {
         assertThat(field.name(), is("y"));
         assertThat(field, instanceOf(Alias.class));
         alias = (Alias) field;
-        assertThat(alias.child().fold(FoldContext.small()), is("2"));
+        assertThat(alias.child().fold(FoldContext.small()), is(BytesRefs.toBytesRef("2")));
 
         field = row.fields().get(2);
         assertThat(field.name(), is("a"));
         assertThat(field, instanceOf(Alias.class));
         alias = (Alias) field;
-        assertThat(alias.child().fold(FoldContext.small()), is("2 days"));
+        assertThat(alias.child().fold(FoldContext.small()), is(BytesRefs.toBytesRef("2 days")));
 
         field = row.fields().get(3);
         assertThat(field.name(), is("b"));
         assertThat(field, instanceOf(Alias.class));
         alias = (Alias) field;
-        assertThat(alias.child().fold(FoldContext.small()), is("4 hours"));
+        assertThat(alias.child().fold(FoldContext.small()), is(BytesRefs.toBytesRef("4 hours")));
 
         field = row.fields().get(4);
         assertThat(field.name(), is("c"));
         assertThat(field, instanceOf(Alias.class));
         alias = (Alias) field;
-        assertThat(alias.child().fold(FoldContext.small()).getClass(), is(String.class));
-        assertThat(alias.child().fold(FoldContext.small()).toString(), is("1.2.3"));
+        assertThat(alias.child().fold(FoldContext.small()).getClass(), is(BytesRef.class));
+        assertThat(alias.child().fold(FoldContext.small()), is(BytesRefs.toBytesRef("1.2.3")));
 
         field = row.fields().get(5);
         assertThat(field.name(), is("d"));
         assertThat(field, instanceOf(Alias.class));
         alias = (Alias) field;
-        assertThat(alias.child().fold(FoldContext.small()).getClass(), is(String.class));
-        assertThat(alias.child().fold(FoldContext.small()).toString(), is("127.0.0.1"));
+        assertThat(alias.child().fold(FoldContext.small()).getClass(), is(BytesRef.class));
+        assertThat(alias.child().fold(FoldContext.small()), is(BytesRefs.toBytesRef("127.0.0.1")));
 
         field = row.fields().get(6);
         assertThat(field.name(), is("e"));
@@ -1659,7 +1661,7 @@ public void testParamInAggFunction() {
         );
         assertThat(plan, instanceOf(Aggregate.class));
         Aggregate agg = (Aggregate) plan;
-        assertThat(((Literal) agg.aggregates().get(0).children().get(0).children().get(0)).value(), equalTo("*"));
+        assertThat(((Literal) agg.aggregates().get(0).children().get(0).children().get(0)).value(), equalTo(BytesRefs.toBytesRef("*")));
         assertThat(agg.child(), instanceOf(Eval.class));
         assertThat(agg.children().size(), equalTo(1));
         assertThat(agg.children().get(0), instanceOf(Eval.class));
@@ -1679,7 +1681,7 @@ public void testParamInAggFunction() {
         );
         assertThat(plan, instanceOf(Aggregate.class));
         agg = (Aggregate) plan;
-        assertThat(((Literal) agg.aggregates().get(0).children().get(0).children().get(0)).value(), equalTo("*"));
+        assertThat(((Literal) agg.aggregates().get(0).children().get(0).children().get(0)).value(), equalTo(BytesRefs.toBytesRef("*")));
         assertThat(agg.child(), instanceOf(Eval.class));
         assertThat(agg.children().size(), equalTo(1));
         assertThat(agg.children().get(0), instanceOf(Eval.class));
@@ -1699,7 +1701,7 @@ public void testParamInAggFunction() {
         );
         assertThat(plan, instanceOf(Aggregate.class));
         agg = (Aggregate) plan;
-        assertThat(((Literal) agg.aggregates().get(0).children().get(0).children().get(0)).value(), equalTo("*"));
+        assertThat(((Literal) agg.aggregates().get(0).children().get(0).children().get(0)).value(), equalTo(BytesRefs.toBytesRef("*")));
         assertThat(agg.child(), instanceOf(Eval.class));
         assertThat(agg.children().size(), equalTo(1));
         assertThat(agg.children().get(0), instanceOf(Eval.class));
@@ -1717,7 +1719,7 @@ public void testParamInAggFunction() {
         );
         assertThat(plan, instanceOf(Aggregate.class));
         agg = (Aggregate) plan;
-        assertThat(((Literal) agg.aggregates().get(0).children().get(0).children().get(0)).value(), equalTo("*"));
+        assertThat(((Literal) agg.aggregates().get(0).children().get(0).children().get(0)).value(), equalTo(BytesRefs.toBytesRef("*")));
         assertThat(agg.child(), instanceOf(Eval.class));
         assertThat(agg.children().size(), equalTo(1));
         assertThat(agg.children().get(0), instanceOf(Eval.class));
@@ -1735,7 +1737,7 @@ public void testParamInAggFunction() {
         );
         assertThat(plan, instanceOf(Aggregate.class));
         agg = (Aggregate) plan;
-        assertThat(((Literal) agg.aggregates().get(0).children().get(0).children().get(0)).value(), equalTo("*"));
+        assertThat(((Literal) agg.aggregates().get(0).children().get(0).children().get(0)).value(), equalTo(BytesRefs.toBytesRef("*")));
         assertThat(agg.child(), instanceOf(Eval.class));
         assertThat(agg.children().size(), equalTo(1));
         assertThat(agg.children().get(0), instanceOf(Eval.class));
@@ -1796,8 +1798,14 @@ public void testIntervalParam() {
         NamedExpression field = eval.fields().get(0);
         assertThat(field.name(), is("y"));
         assertThat(field, instanceOf(Alias.class));
-        assertThat(((Literal) ((Add) eval.fields().get(0).child()).left().children().get(0)).value(), equalTo("2024-01-01"));
-        assertThat(((Literal) ((Add) eval.fields().get(0).child()).right().children().get(0)).value(), equalTo("3 days"));
+        assertThat(
+            ((Literal) ((Add) eval.fields().get(0).child()).left().children().get(0)).value(),
+            equalTo(BytesRefs.toBytesRef("2024-01-01"))
+        );
+        assertThat(
+            ((Literal) ((Add) eval.fields().get(0).child()).right().children().get(0)).value(),
+            equalTo(BytesRefs.toBytesRef("3 days"))
+        );
     }
 
     public void testParamForIdentifier() {
@@ -2017,7 +2025,7 @@ public void testParamForIdentifier() {
                 EMPTY,
                 relation("idx1"),
                 null,
-                new Literal(EMPTY, "idx2", KEYWORD),
+                Literal.keyword(EMPTY, "idx2"),
                 attribute("f.1.*"),
                 null,
                 Map.of(),
@@ -2034,7 +2042,7 @@ public void testParamForIdentifier() {
                 EMPTY,
                 relation("idx1"),
                 null,
-                new Literal(EMPTY, "idx2", KEYWORD),
+                Literal.keyword(EMPTY, "idx2"),
                 attribute("f.1.*.f.2"),
                 null,
                 Map.of(),
@@ -2273,7 +2281,7 @@ private void assertStringAsLookupIndexPattern(String string, String statement) {
         var plan = statement(statement);
         var lookup = as(plan, Lookup.class);
         var tableName = as(lookup.tableName(), Literal.class);
-        assertThat(tableName.fold(FoldContext.small()), equalTo(string));
+        assertThat(tableName.fold(FoldContext.small()), equalTo(BytesRefs.toBytesRef(string)));
     }
 
     public void testIdPatternUnquoted() {
@@ -2344,7 +2352,7 @@ public void testLookup() {
         var plan = statement(query);
         var lookup = as(plan, Lookup.class);
         var tableName = as(lookup.tableName(), Literal.class);
-        assertThat(tableName.fold(FoldContext.small()), equalTo("t"));
+        assertThat(tableName.fold(FoldContext.small()), equalTo(BytesRefs.toBytesRef("t")));
         assertThat(lookup.matchFields(), hasSize(1));
         var matchField = as(lookup.matchFields().get(0), UnresolvedAttribute.class);
         assertThat(matchField.name(), equalTo("j"));
@@ -2526,7 +2534,7 @@ public void testMatchOperatorConstantQueryString() {
         var match = (MatchOperator) filter.condition();
         var matchField = (UnresolvedAttribute) match.field();
         assertThat(matchField.name(), equalTo("field"));
-        assertThat(match.query().fold(FoldContext.small()), equalTo("value"));
+        assertThat(match.query().fold(FoldContext.small()), equalTo(BytesRefs.toBytesRef("value")));
     }
 
     public void testInvalidMatchOperator() {
@@ -2561,7 +2569,7 @@ public void testMatchFunctionFieldCasting() {
         var toInteger = (ToInteger) function.children().get(0);
         var matchField = (UnresolvedAttribute) toInteger.field();
         assertThat(matchField.name(), equalTo("field"));
-        assertThat(function.children().get(1).fold(FoldContext.small()), equalTo("value"));
+        assertThat(function.children().get(1).fold(FoldContext.small()), equalTo(BytesRefs.toBytesRef("value")));
     }
 
     public void testMatchOperatorFieldCasting() {
@@ -2571,7 +2579,7 @@ public void testMatchOperatorFieldCasting() {
         var toInteger = (ToInteger) match.field();
         var matchField = (UnresolvedAttribute) toInteger.field();
         assertThat(matchField.name(), equalTo("field"));
-        assertThat(match.query().fold(FoldContext.small()), equalTo("value"));
+        assertThat(match.query().fold(FoldContext.small()), equalTo(BytesRefs.toBytesRef("value")));
     }
 
     public void testFailingMetadataWithSquareBrackets() {
@@ -2612,17 +2620,14 @@ public void testNamedFunctionArgumentInMap() {
                         new Alias(
                             EMPTY,
                             "x",
-                            function(
-                                "fn1",
-                                List.of(attribute("f1"), new Literal(EMPTY, "testString", KEYWORD), mapExpression(expectedMap1))
-                            )
+                            function("fn1", List.of(attribute("f1"), Literal.keyword(EMPTY, "testString"), mapExpression(expectedMap1)))
                         )
                     )
                 ),
                 new Equals(
                     EMPTY,
                     attribute("y"),
-                    function("fn2", List.of(new Literal(EMPTY, "testString", KEYWORD), mapExpression(expectedMap2)))
+                    function("fn2", List.of(Literal.keyword(EMPTY, "testString"), mapExpression(expectedMap2)))
                 )
             ),
             statement("""
@@ -2714,17 +2719,14 @@ public void testNamedFunctionArgumentInMapWithNamedParameters() {
                         new Alias(
                             EMPTY,
                             "x",
-                            function(
-                                "fn1",
-                                List.of(attribute("f1"), new Literal(EMPTY, "testString", KEYWORD), mapExpression(expectedMap1))
-                            )
+                            function("fn1", List.of(attribute("f1"), Literal.keyword(EMPTY, "testString"), mapExpression(expectedMap1)))
                         )
                     )
                 ),
                 new Equals(
                     EMPTY,
                     attribute("y"),
-                    function("fn2", List.of(new Literal(EMPTY, "testString", KEYWORD), mapExpression(expectedMap2)))
+                    function("fn2", List.of(Literal.keyword(EMPTY, "testString"), mapExpression(expectedMap2)))
                 )
             ),
             statement(
@@ -2853,17 +2855,14 @@ public void testNamedFunctionArgumentWithCaseSensitiveKeys() {
                         new Alias(
                             EMPTY,
                             "x",
-                            function(
-                                "fn1",
-                                List.of(attribute("f1"), new Literal(EMPTY, "testString", KEYWORD), mapExpression(expectedMap1))
-                            )
+                            function("fn1", List.of(attribute("f1"), Literal.keyword(EMPTY, "testString"), mapExpression(expectedMap1)))
                         )
                     )
                 ),
                 new Equals(
                     EMPTY,
                     attribute("y"),
-                    function("fn2", List.of(new Literal(EMPTY, "testString", KEYWORD), mapExpression(expectedMap2)))
+                    function("fn2", List.of(Literal.keyword(EMPTY, "testString"), mapExpression(expectedMap2)))
                 )
             ),
             statement("""
@@ -3303,7 +3302,7 @@ public void testValidFork() {
         var match = (MatchOperator) filter.condition();
         var matchField = (UnresolvedAttribute) match.field();
         assertThat(matchField.name(), equalTo("a"));
-        assertThat(match.query().fold(FoldContext.small()), equalTo("baz"));
+        assertThat(match.query().fold(FoldContext.small()), equalTo(BytesRefs.toBytesRef("baz")));
 
         // second subplan
         eval = as(subPlans.get(1), Eval.class);
@@ -3317,7 +3316,7 @@ public void testValidFork() {
         match = (MatchOperator) filter.condition();
         matchField = (UnresolvedAttribute) match.field();
         assertThat(matchField.name(), equalTo("b"));
-        assertThat(match.query().fold(FoldContext.small()), equalTo("bar"));
+        assertThat(match.query().fold(FoldContext.small()), equalTo(BytesRefs.toBytesRef("bar")));
 
         // third subplan
         eval = as(subPlans.get(2), Eval.class);
@@ -3326,7 +3325,7 @@ public void testValidFork() {
         match = (MatchOperator) filter.condition();
         matchField = (UnresolvedAttribute) match.field();
         assertThat(matchField.name(), equalTo("c"));
-        assertThat(match.query().fold(FoldContext.small()), equalTo("bat"));
+        assertThat(match.query().fold(FoldContext.small()), equalTo(BytesRefs.toBytesRef("bat")));
 
         // fourth subplan
         eval = as(subPlans.get(3), Eval.class);
@@ -4057,7 +4056,7 @@ public void testDoubleParamsForIdentifier() {
                     EMPTY,
                     relation("idx1"),
                     null,
-                    new Literal(EMPTY, "idx2", KEYWORD),
+                    Literal.keyword(EMPTY, "idx2"),
                     attribute("f.1.*"),
                     null,
                     Map.of(),
@@ -4093,7 +4092,7 @@ public void testDoubleParamsForIdentifier() {
                     EMPTY,
                     relation("idx1"),
                     null,
-                    new Literal(EMPTY, "idx2", KEYWORD),
+                    Literal.keyword(EMPTY, "idx2"),
                     attribute("f.1.*.f.2"),
                     null,
                     Map.of(),
@@ -4145,7 +4144,7 @@ public void testMixedSingleDoubleParams() {
                         new Eval(
                             EMPTY,
                             relation("test"),
-                            List.of(new Alias(EMPTY, "x", function("toString", List.of(new Literal(EMPTY, "constant_value", KEYWORD)))))
+                            List.of(new Alias(EMPTY, "x", function("toString", List.of(Literal.keyword(EMPTY, "constant_value")))))
                         ),
                         new Equals(EMPTY, attribute("f.2"), new Literal(EMPTY, 100, INTEGER))
                     )
@@ -4188,7 +4187,7 @@ public void testMixedSingleDoubleParams() {
                             EMPTY,
                             relation("test"),
                             List.of(attribute("f.4.")),
-                            List.of(new Alias(EMPTY, "y", function("count", List.of(new Literal(EMPTY, "*", KEYWORD)))), attribute("f.4."))
+                            List.of(new Alias(EMPTY, "y", function("count", List.of(Literal.keyword(EMPTY, "*")))), attribute("f.4."))
                         ),
                         List.of(new Order(EMPTY, attribute("f.5.*"), Order.OrderDirection.ASC, Order.NullsPosition.LAST))
                     ),
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/AggregateSerializationTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/AggregateSerializationTests.java
index 055b094dd8a95..440b63050c418 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/AggregateSerializationTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/AggregateSerializationTests.java
@@ -51,7 +51,7 @@ public static List<? extends NamedExpression> randomAggregates() {
                     randomSource(),
                     FieldAttributeTests.createFieldAttribute(1, true),
                     new Literal(randomSource(), between(1, 5), DataType.INTEGER),
-                    new Literal(randomSource(), randomFrom("ASC", "DESC"), DataType.KEYWORD)
+                    Literal.keyword(randomSource(), randomFrom("ASC", "DESC"))
                 );
                 case 4 -> new Values(randomSource(), FieldAttributeTests.createFieldAttribute(1, true));
                 case 5 -> new Sum(randomSource(), FieldAttributeTests.createFieldAttribute(1, true));
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/EnrichSerializationTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/EnrichSerializationTests.java
index 52c2035702354..1f27dd1a4522f 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/EnrichSerializationTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/EnrichSerializationTests.java
@@ -20,7 +20,6 @@
 import org.elasticsearch.xpack.esql.core.expression.Literal;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.core.tree.Source;
-import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.expression.function.FieldAttributeTests;
 
 import java.io.ByteArrayOutputStream;
@@ -44,7 +43,7 @@ protected Enrich createTestInstance() {
     }
 
     private static Expression randomPolicyName() {
-        return new Literal(randomSource(), randomAlphaOfLength(5), DataType.KEYWORD);
+        return Literal.keyword(randomSource(), randomAlphaOfLength(5));
     }
 
     private static EnrichPolicy randomEnrichPolicy() {
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/inference/CompletionSerializationTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/inference/CompletionSerializationTests.java
index 0b6c1f4eb1b2c..e9810454224aa 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/inference/CompletionSerializationTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/inference/CompletionSerializationTests.java
@@ -11,7 +11,6 @@
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
 import org.elasticsearch.xpack.esql.core.tree.Source;
-import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.expression.function.ReferenceAttributeTests;
 import org.elasticsearch.xpack.esql.plan.logical.AbstractLogicalPlanSerializationTests;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
@@ -42,11 +41,11 @@ protected Completion mutateInstance(Completion instance) throws IOException {
     }
 
     private Literal randomInferenceId() {
-        return new Literal(Source.EMPTY, randomIdentifier(), DataType.KEYWORD);
+        return Literal.keyword(Source.EMPTY, randomIdentifier());
     }
 
     private Expression randomPrompt() {
-        return randomBoolean() ? new Literal(Source.EMPTY, randomIdentifier(), DataType.KEYWORD) : randomAttribute();
+        return randomBoolean() ? Literal.keyword(Source.EMPTY, randomIdentifier()) : randomAttribute();
     }
 
     private Attribute randomAttribute() {
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/inference/RerankSerializationTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/inference/RerankSerializationTests.java
index 22f60c78ec842..f5819f7767e7a 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/inference/RerankSerializationTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/inference/RerankSerializationTests.java
@@ -52,7 +52,7 @@ private List<Alias> randomFields() {
     }
 
     private Literal string(String value) {
-        return new Literal(EMPTY, value, DataType.KEYWORD);
+        return Literal.keyword(EMPTY, value);
     }
 
     private Attribute scoreAttribute() {
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/physical/inference/CompletionExecSerializationTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/physical/inference/CompletionExecSerializationTests.java
index 92d2a4b445c6a..9fd41a2432462 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/physical/inference/CompletionExecSerializationTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/physical/inference/CompletionExecSerializationTests.java
@@ -11,7 +11,6 @@
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
 import org.elasticsearch.xpack.esql.core.tree.Source;
-import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.expression.function.ReferenceAttributeTests;
 import org.elasticsearch.xpack.esql.plan.physical.AbstractPhysicalPlanSerializationTests;
 import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
@@ -41,11 +40,11 @@ protected CompletionExec mutateInstance(CompletionExec instance) throws IOExcept
     }
 
     private Literal randomInferenceId() {
-        return new Literal(Source.EMPTY, randomIdentifier(), DataType.KEYWORD);
+        return Literal.keyword(Source.EMPTY, randomIdentifier());
     }
 
     private Expression randomPrompt() {
-        return randomBoolean() ? new Literal(Source.EMPTY, randomIdentifier(), DataType.KEYWORD) : randomAttribute();
+        return randomBoolean() ? Literal.keyword(Source.EMPTY, randomIdentifier()) : randomAttribute();
     }
 
     private Attribute randomAttribute() {
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/physical/inference/RerankExecSerializationTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/physical/inference/RerankExecSerializationTests.java
index f5ba1718c7ea0..fba846ac02145 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/physical/inference/RerankExecSerializationTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/physical/inference/RerankExecSerializationTests.java
@@ -52,7 +52,7 @@ private List<Alias> randomFields() {
     }
 
     static Literal string(String value) {
-        return new Literal(EMPTY, value, DataType.KEYWORD);
+        return Literal.keyword(EMPTY, value);
     }
 
     private Attribute scoreAttribute() {
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/type/DataTypeConversionTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/type/DataTypeConversionTests.java
index 871bf632adcc6..16935afcc0ef6 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/type/DataTypeConversionTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/type/DataTypeConversionTests.java
@@ -6,6 +6,7 @@
  */
 package org.elasticsearch.xpack.esql.type;
 
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.esql.core.InvalidArgumentException;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
@@ -545,7 +546,7 @@ public void testIpToString() {
         Converter ipToString = converterFor(IP, KEYWORD);
         assertEquals("10.0.0.1", ipToString.convert(new Literal(s, "10.0.0.1", IP)));
         Converter stringToIp = converterFor(KEYWORD, IP);
-        assertEquals("10.0.0.1", ipToString.convert(stringToIp.convert(new Literal(s, "10.0.0.1", KEYWORD))));
+        assertEquals("10.0.0.1", ipToString.convert(stringToIp.convert(Literal.keyword(s, "10.0.0.1"))));
     }
 
     public void testStringToVersion() {
@@ -562,10 +563,13 @@ public void testVersionToString() {
         Source s2 = new Source(Location.EMPTY, "2.1.4-SNAPSHOT");
         DataType stringType = randomFrom(TEXT, KEYWORD);
         Converter versionToString = converterFor(VERSION, stringType);
-        assertEquals("2.1.4", versionToString.convert(new Literal(s, "2.1.4", VERSION)));
-        assertEquals("2.1.4-SNAPSHOT", versionToString.convert(new Literal(s2, "2.1.4-SNAPSHOT", VERSION)));
+        assertEquals("2.1.4", versionToString.convert(new Literal(s, new Version("2.1.4").toBytesRef(), VERSION)));
+        assertEquals("2.1.4-SNAPSHOT", versionToString.convert(new Literal(s2, new Version("2.1.4-SNAPSHOT").toBytesRef(), VERSION)));
         Converter stringToVersion = converterFor(stringType, VERSION);
-        assertEquals("2.1.4", versionToString.convert(stringToVersion.convert(new Literal(s, "2.1.4", stringType))));
-        assertEquals("2.1.4-SNAPSHOT", versionToString.convert(stringToVersion.convert(new Literal(s2, "2.1.4-SNAPSHOT", stringType))));
+        assertEquals("2.1.4", versionToString.convert(stringToVersion.convert(new Literal(s, BytesRefs.toBytesRef("2.1.4"), stringType))));
+        assertEquals(
+            "2.1.4-SNAPSHOT",
+            versionToString.convert(stringToVersion.convert(new Literal(s2, BytesRefs.toBytesRef("2.1.4-SNAPSHOT"), stringType)))
+        );
     }
 }