[9.1] Fix behavior for _index LIKE for ESQL (#130849) (#130948)

* Fix behavior for _index LIKE for ESQL (#130849)

Fixes _index LIKE <pattern> to always have normal text matching semantics.

Implement a generic ExpressionQuery and ExpressionQueryBuilder that can be serialized to the data node. Then the ExpressionQueryBuilder can build an Automaton using TranslationAware.asLuceneQuery() and execute it in Lucine.

Introduces a breaking change for LIKE on _index fields. The old like behavior is not correct and does not have normal like semantics from ESQL. Customers upgrading from old build to new build might see a regression, where the data changes due to the like filters on clustering produces different results, but the new results are correct.

Behavior for ESQL
New CCS to New => New behavior everywhere
Old CCS to New => Old behavior everywhere (the isForESQL flag is not passed in from old)
New CCS to Old => New behavior for new, old behavior for old (the isForESQL cannot be passed, old does not know about it).
Old CCS to Old => Old behavior everywhere

Closes #129511

(cherry picked from commit 8c4eaf957dcf46a269dacf8b9be8d3aa0cb422f0)

# Conflicts:
#	server/src/main/java/org/elasticsearch/TransportVersions.java
#	x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java
#	x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LocalPhysicalPlanOptimizerTests.java

* Prepare for backport to 9.1 and 8.19

* Address code review feedback

* Transport version backport

* [CI] Auto commit changes from spotless

* Fix merge errors

---------

Co-authored-by: elasticsearchmachine <infra-root+elasticsearchmachine@elastic.co>

benchmarks/src/main/java/org/elasticsearch/benchmark/_nightly/esql/QueryPlanningBenchmark.java

@@ -70,11 +70,11 @@ public class QueryPlanningBenchmark {
     private EsqlParser defaultParser;
     private Analyzer manyFieldsAnalyzer;
     private LogicalPlanOptimizer defaultOptimizer;
+    private Configuration config;
 
     @Setup
     public void setup() {
-
-        var config = new Configuration(
+        this.config = new Configuration(
             DateUtils.UTC,
             Locale.US,
             null,
@@ -116,7 +116,7 @@ public class QueryPlanningBenchmark {
     }
 
     private LogicalPlan plan(EsqlParser parser, Analyzer analyzer, LogicalPlanOptimizer optimizer, String query) {
-        var parsed = parser.createStatement(query, new QueryParams(), telemetry);
+        var parsed = parser.createStatement(query, new QueryParams(), telemetry, config);
         var analyzed = analyzer.analyze(parsed);
         var optimized = optimizer.optimize(analyzed);
         return optimized;

docs/changelog/130849.yaml

@@ -0,0 +1,6 @@
+pr: 130849
+summary: Fix behavior for `_index` LIKE for ESQL
+area: ES|QL
+type: bug
+issues:
+ - 129511

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -211,6 +211,7 @@ public class TransportVersions {
     public static final TransportVersion ESQL_DOCUMENTS_FOUND_AND_VALUES_LOADED_8_19 = def(8_841_0_61);
     public static final TransportVersion ESQL_PROFILE_INCLUDE_PLAN_8_19 = def(8_841_0_62);
     public static final TransportVersion ESQL_SPLIT_ON_BIG_VALUES_8_19 = def(8_841_0_63);
+    public static final TransportVersion ESQL_FIXED_INDEX_LIKE_8_19 = def(8_841_0_64);
     public static final TransportVersion V_9_0_0 = def(9_000_0_09);
     public static final TransportVersion INITIAL_ELASTICSEARCH_9_0_1 = def(9_000_0_10);
     public static final TransportVersion INITIAL_ELASTICSEARCH_9_0_2 = def(9_000_0_11);
@@ -328,6 +329,7 @@ public class TransportVersions {
     public static final TransportVersion ESQL_PROFILE_INCLUDE_PLAN = def(9_111_0_00);
     public static final TransportVersion MAPPINGS_IN_DATA_STREAMS = def(9_112_0_00);
     public static final TransportVersion ESQL_SPLIT_ON_BIG_VALUES_9_1 = def(9_112_0_01);
+    public static final TransportVersion ESQL_FIXED_INDEX_LIKE_9_1 = def(9_112_0_02);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/index/mapper/ConstantFieldType.java

@@ -15,6 +15,8 @@ import org.apache.lucene.search.MatchNoDocsQuery;
 import org.apache.lucene.search.MultiTermQuery;
 import org.apache.lucene.search.Query;
 import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.automaton.Automaton;
+import org.apache.lucene.util.automaton.CharacterRunAutomaton;
 import org.elasticsearch.common.lucene.search.Queries;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.core.Nullable;
@@ -23,6 +25,7 @@ import org.elasticsearch.index.query.SearchExecutionContext;
 
 import java.util.Collection;
 import java.util.Map;
+import java.util.function.Supplier;
 
 /**
  * A {@link MappedFieldType} that has the same value for all documents.
@@ -135,9 +138,47 @@ public abstract class ConstantFieldType extends MappedFieldType {
         }
     }
 
+    /**
+     * Returns a query that matches all documents or no documents
+     * It usually calls {@link #wildcardQuery(String, boolean, QueryRewriteContext)}
+     * except for IndexFieldType which overrides this method to use its own matching logic.
+     */
+    public Query wildcardLikeQuery(String value, boolean caseInsensitive, QueryRewriteContext context) {
+        return wildcardQuery(value, caseInsensitive, context);
+    }
+
     @Override
     public final boolean fieldHasValue(FieldInfos fieldInfos) {
         // We consider constant field types to always have value.
         return true;
     }
+
+    /**
+     * Returns the constant value of this field as a string.
+     * Based on the field type, we need to get it in a different way.
+     */
+    public abstract String getConstantFieldValue(SearchExecutionContext context);
+
+    /**
+     * Returns a query that matches all documents or no documents
+     * depending on whether the constant value of this field matches or not
+     */
+    @Override
+    public Query automatonQuery(
+        Supplier<Automaton> automatonSupplier,
+        Supplier<CharacterRunAutomaton> characterRunAutomatonSupplier,
+        @Nullable MultiTermQuery.RewriteMethod method,
+        SearchExecutionContext context,
+        String description
+    ) {
+        CharacterRunAutomaton compiled = characterRunAutomatonSupplier.get();
+        boolean matches = compiled.run(getConstantFieldValue(context));
+        if (matches) {
+            return new MatchAllDocsQuery();
+        } else {
+            return new MatchNoDocsQuery(
+                "The \"" + context.getFullyQualifiedIndex().getName() + "\" query was rewritten to a \"match_none\" query."
+            );
+        }
+    }
 }

server/src/main/java/org/elasticsearch/index/mapper/IndexFieldMapper.java

@@ -10,9 +10,13 @@
 package org.elasticsearch.index.mapper;
 
 import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.search.MatchNoDocsQuery;
+import org.apache.lucene.search.MultiTermQuery;
 import org.apache.lucene.search.Query;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.regex.Regex;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.fielddata.FieldData;
 import org.elasticsearch.index.fielddata.FieldDataContext;
 import org.elasticsearch.index.fielddata.IndexFieldData;
@@ -27,6 +31,7 @@ import org.elasticsearch.search.lookup.Source;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Locale;
 
 public class IndexFieldMapper extends MetadataFieldMapper {
 
@@ -102,6 +107,38 @@ public class IndexFieldMapper extends MetadataFieldMapper {
             };
         }
 
+        @Override
+        public Query wildcardLikeQuery(
+            String value,
+            @Nullable MultiTermQuery.RewriteMethod method,
+            boolean caseInsensitve,
+            SearchExecutionContext context
+        ) {
+            String indexName = context.getFullyQualifiedIndex().getName();
+            return getWildcardLikeQuery(value, caseInsensitve, indexName);
+        }
+
+        @Override
+        public Query wildcardLikeQuery(String value, boolean caseInsensitive, QueryRewriteContext context) {
+            String indexName = context.getFullyQualifiedIndex().getName();
+            return getWildcardLikeQuery(value, caseInsensitive, indexName);
+        }
+
+        private static Query getWildcardLikeQuery(String value, boolean caseInsensitve, String indexName) {
+            if (caseInsensitve) {
+                value = value.toLowerCase(Locale.ROOT);
+                indexName = indexName.toLowerCase(Locale.ROOT);
+            }
+            if (Regex.simpleMatch(value, indexName)) {
+                return new MatchAllDocsQuery();
+            }
+            return new MatchNoDocsQuery("The \"" + indexName + "\" query was rewritten to a \"match_none\" query.");
+        }
+
+        @Override
+        public String getConstantFieldValue(SearchExecutionContext context) {
+            return context.getFullyQualifiedIndex().getName();
+        }
     }
 
     public IndexFieldMapper() {

server/src/main/java/org/elasticsearch/index/mapper/IndexModeFieldMapper.java

@@ -57,6 +57,11 @@ public class IndexModeFieldMapper extends MetadataFieldMapper {
             return Regex.simpleMatch(pattern, indexMode, caseInsensitive);
         }
 
+        @Override
+        public String getConstantFieldValue(SearchExecutionContext context) {
+            return context.getIndexSettings().getMode().getName();
+        }
+
         @Override
         public Query existsQuery(SearchExecutionContext context) {
             return new MatchAllDocsQuery();

server/src/main/java/org/elasticsearch/index/mapper/KeywordFieldMapper.java

@@ -24,6 +24,7 @@ import org.apache.lucene.index.IndexOptions;
 import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.MultiTerms;
+import org.apache.lucene.index.Term;
 import org.apache.lucene.index.Terms;
 import org.apache.lucene.index.TermsEnum;
 import org.apache.lucene.search.MultiTermQuery;
@@ -31,6 +32,7 @@ import org.apache.lucene.search.Query;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.automaton.Automata;
 import org.apache.lucene.util.automaton.Automaton;
+import org.apache.lucene.util.automaton.CharacterRunAutomaton;
 import org.apache.lucene.util.automaton.CompiledAutomaton;
 import org.apache.lucene.util.automaton.CompiledAutomaton.AUTOMATON_TYPE;
 import org.apache.lucene.util.automaton.Operations;
@@ -51,6 +53,7 @@ import org.elasticsearch.index.fielddata.IndexFieldData;
 import org.elasticsearch.index.fielddata.SourceValueFetcherSortedBinaryIndexFieldData;
 import org.elasticsearch.index.fielddata.StoredFieldSortedBinaryIndexFieldData;
 import org.elasticsearch.index.fielddata.plain.SortedSetOrdinalsIndexFieldData;
+import org.elasticsearch.index.query.AutomatonQueryWithDescription;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.index.similarity.SimilarityProvider;
 import org.elasticsearch.script.Script;
@@ -82,6 +85,7 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.Supplier;
 
 import static org.apache.lucene.index.IndexWriter.MAX_TERM_LENGTH;
 import static org.elasticsearch.core.Strings.format;
@@ -1042,6 +1046,17 @@ public final class KeywordFieldMapper extends FieldMapper {
         public boolean hasDocValuesSkipper() {
             return hasDocValuesSkipper;
         }
+
+        @Override
+        public Query automatonQuery(
+            Supplier<Automaton> automatonSupplier,
+            Supplier<CharacterRunAutomaton> characterRunAutomatonSupplier,
+            @Nullable MultiTermQuery.RewriteMethod method,
+            SearchExecutionContext context,
+            String description
+        ) {
+            return new AutomatonQueryWithDescription(new Term(name()), automatonSupplier.get(), description);
+        }
     }
 
     private final boolean indexed;

server/src/main/java/org/elasticsearch/index/mapper/MappedFieldType.java

@@ -25,6 +25,8 @@ import org.apache.lucene.search.MultiTermQuery;
 import org.apache.lucene.search.Query;
 import org.apache.lucene.search.TermQuery;
 import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.automaton.Automaton;
+import org.apache.lucene.util.automaton.CharacterRunAutomaton;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
@@ -54,6 +56,7 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 import static org.elasticsearch.search.SearchService.ALLOW_EXPENSIVE_QUERIES;
 
@@ -329,6 +332,19 @@ public abstract class MappedFieldType {
         return wildcardQuery(value, method, false, context);
     }
 
+    /**
+     * Similar to wildcardQuery, except that we change the behavior for ESQL
+     * to behave like a string LIKE query, where the value is matched as a string
+     */
+    public Query wildcardLikeQuery(
+        String value,
+        @Nullable MultiTermQuery.RewriteMethod method,
+        boolean caseInsensitve,
+        SearchExecutionContext context
+    ) {
+        return wildcardQuery(value, method, caseInsensitve, context);
+    }
+
     public Query wildcardQuery(
         String value,
         @Nullable MultiTermQuery.RewriteMethod method,
@@ -370,6 +386,23 @@ public abstract class MappedFieldType {
         );
     }
 
+    /**
+     * Returns a Lucine pushable Query for the current field
+     * For now can only be AutomatonQuery or MatchAllDocsQuery() or MatchNoDocsQuery()
+     */
+    public Query automatonQuery(
+        Supplier<Automaton> automatonSupplier,
+        Supplier<CharacterRunAutomaton> characterRunAutomatonSupplier,
+        @Nullable MultiTermQuery.RewriteMethod method,
+        SearchExecutionContext context,
+        String description
+    ) {
+        throw new QueryShardException(
+            context,
+            "Can only use automaton queries on keyword fields - not on [" + name + "] which is of type [" + typeName() + "]"
+        );
+    }
+
     public Query existsQuery(SearchExecutionContext context) {
         if (hasDocValues() || getTextSearchInfo().hasNorms()) {
             return new FieldExistsQuery(name());

server/src/main/java/org/elasticsearch/index/query/AutomatonQueryBuilder.java

@@ -1,108 +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", the "GNU Affero General Public License v3.0 only", and the "Server Side
- * Public License v 1"; you may not use this file except in compliance with, at
- * your election, the "Elastic License 2.0", the "GNU Affero General Public
- * License v3.0 only", or the "Server Side Public License, v 1".
- */
-
-package org.elasticsearch.index.query;
-
-import org.apache.lucene.index.Term;
-import org.apache.lucene.search.AutomatonQuery;
-import org.apache.lucene.search.Query;
-import org.apache.lucene.util.automaton.Automaton;
-import org.elasticsearch.TransportVersion;
-import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.xcontent.XContentBuilder;
-
-import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.util.Objects;
-
-/**
- * Implements an Automaton query, which matches documents based on a Lucene Automaton.
- * It does not support serialization or XContent representation.
- */
-public class AutomatonQueryBuilder extends AbstractQueryBuilder<AutomatonQueryBuilder> implements MultiTermQueryBuilder {
-    private final String fieldName;
-    private final Automaton automaton;
-    private final String description;
-
-    public AutomatonQueryBuilder(String fieldName, Automaton automaton, String description) {
-        if (Strings.isEmpty(fieldName)) {
-            throw new IllegalArgumentException("field name is null or empty");
-        }
-        if (automaton == null) {
-            throw new IllegalArgumentException("automaton cannot be null");
-        }
-        this.fieldName = fieldName;
-        this.automaton = automaton;
-        this.description = description;
-    }
-
-    @Override
-    public String fieldName() {
-        return fieldName;
-    }
-
-    public String description() {
-        return description;
-    }
-
-    @Override
-    public String getWriteableName() {
-        throw new UnsupportedOperationException("AutomatonQueryBuilder does not support getWriteableName");
-    }
-
-    @Override
-    protected void doWriteTo(StreamOutput out) throws IOException {
-        throw new UnsupportedEncodingException("AutomatonQueryBuilder does not support doWriteTo");
-    }
-
-    @Override
-    protected void doXContent(XContentBuilder builder, Params params) throws IOException {
-        throw new UnsupportedEncodingException("AutomatonQueryBuilder does not support doXContent");
-    }
-
-    @Override
-    protected Query doToQuery(SearchExecutionContext context) throws IOException {
-        return new AutomatonQueryWithDescription(new Term(fieldName), automaton, description);
-    }
-
-    @Override
-    protected int doHashCode() {
-        return Objects.hash(fieldName, automaton, description);
-    }
-
-    @Override
-    protected boolean doEquals(AutomatonQueryBuilder other) {
-        return Objects.equals(fieldName, other.fieldName)
-            && Objects.equals(automaton, other.automaton)
-            && Objects.equals(description, other.description);
-    }
-
-    @Override
-    public TransportVersion getMinimalSupportedVersion() {
-        throw new UnsupportedOperationException("AutomatonQueryBuilder does not support getMinimalSupportedVersion");
-    }
-
-    static class AutomatonQueryWithDescription extends AutomatonQuery {
-        private final String description;
-
-        AutomatonQueryWithDescription(Term term, Automaton automaton, String description) {
-            super(term, automaton);
-            this.description = description;
-        }
-
-        @Override
-        public String toString(String field) {
-            if (this.field.equals(field)) {
-                return description;
-            }
-            return this.field + ":" + description;
-        }
-    }
-}

server/src/main/java/org/elasticsearch/index/query/AutomatonQueryWithDescription.java

@@ -0,0 +1,36 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index.query;
+
+import org.apache.lucene.index.Term;
+import org.apache.lucene.search.AutomatonQuery;
+import org.apache.lucene.util.automaton.Automaton;
+
+/**
+ * A specialized {@link AutomatonQuery} that includes a description of the query.
+ * This can be useful for debugging or logging purposes, providing more context
+ * about the query being executed.
+ */
+public class AutomatonQueryWithDescription extends AutomatonQuery {
+    private final String description;
+
+    public AutomatonQueryWithDescription(Term term, Automaton automaton, String description) {
+        super(term, automaton);
+        this.description = description;
+    }
+
+    @Override
+    public String toString(String field) {
+        if (this.field.equals(field)) {
+            return description;
+        }
+        return this.field + ":" + description;
+    }
+}

server/src/main/java/org/elasticsearch/index/query/CoordinatorRewriteContext.java

@@ -72,6 +72,11 @@ public class CoordinatorRewriteContext extends QueryRewriteContext {
             return Regex.simpleMatch(pattern, tierPreference);
         }
 
+        @Override
+        public String getConstantFieldValue(SearchExecutionContext context) {
+            return context.getTierPreference();
+        }
+
         @Override
         public Query existsQuery(SearchExecutionContext context) {
             throw new UnsupportedOperationException("field exists query is not supported on the coordinator node");

server/src/main/java/org/elasticsearch/index/query/WildcardQueryBuilder.java

@@ -56,6 +56,13 @@ public class WildcardQueryBuilder extends AbstractQueryBuilder<WildcardQueryBuil
     private static final ParseField CASE_INSENSITIVE_FIELD = new ParseField("case_insensitive");
     private boolean caseInsensitive = DEFAULT_CASE_INSENSITIVITY;
 
+    /**
+     * Force string matching instead of the field-type-aware wildcard matching.
+     * When this is true the {@link org.elasticsearch.index.mapper.IndexFieldMapper} will always match of the
+     * {@code cluster_name:index_name} instead of emulating the glob pattern on the URL.
+     */
+    private boolean forceStringMatch = false;
+
     /**
      * Implements the wildcard search query. Supported wildcards are {@code *}, which
      * matches any character sequence (including the empty one), and {@code ?},
@@ -78,6 +85,16 @@ public class WildcardQueryBuilder extends AbstractQueryBuilder<WildcardQueryBuil
         this.value = value;
     }
 
+    /**
+     * @param forceStringMatch Force string matching instead of the field-type-aware wildcard matching.
+     * When this is true the {@link org.elasticsearch.index.mapper.IndexFieldMapper} will always match of the
+     * {@code cluster_name:index_name} instead of emulating the glob pattern on the URL.
+     */
+    public WildcardQueryBuilder(String fieldName, String value, boolean forceStringMatch) {
+        this(fieldName, value);
+        this.forceStringMatch = forceStringMatch;
+    }
+
     /**
      * Read from a stream.
      */
@@ -87,6 +104,11 @@ public class WildcardQueryBuilder extends AbstractQueryBuilder<WildcardQueryBuil
         value = in.readString();
         rewrite = in.readOptionalString();
         caseInsensitive = in.readBoolean();
+        if (expressionTransportSupported(in.getTransportVersion())) {
+            forceStringMatch = in.readBoolean();
+        } else {
+            forceStringMatch = false;
+        }
     }
 
     @Override
@@ -95,6 +117,17 @@ public class WildcardQueryBuilder extends AbstractQueryBuilder<WildcardQueryBuil
         out.writeString(value);
         out.writeOptionalString(rewrite);
         out.writeBoolean(caseInsensitive);
+        if (expressionTransportSupported(out.getTransportVersion())) {
+            out.writeBoolean(forceStringMatch);
+        }
+    }
+
+    /**
+     * Returns true if the Transport version is compatible with ESQL_FIXED_INDEX_LIKE
+     */
+    public static boolean expressionTransportSupported(TransportVersion version) {
+        return version.onOrAfter(TransportVersions.ESQL_FIXED_INDEX_LIKE_9_1)
+            || version.isPatchFrom(TransportVersions.ESQL_FIXED_INDEX_LIKE_8_19);
     }
 
     @Override
@@ -218,7 +251,12 @@ public class WildcardQueryBuilder extends AbstractQueryBuilder<WildcardQueryBuil
             // This logic is correct for all field types, but by only applying it to constant
             // fields we also have the guarantee that it doesn't perform I/O, which is important
             // since rewrites might happen on a network thread.
-            Query query = constantFieldType.wildcardQuery(value, caseInsensitive, context); // the rewrite method doesn't matter
+            Query query;
+            if (forceStringMatch) {
+                query = constantFieldType.wildcardLikeQuery(value, caseInsensitive, context); // the rewrite method doesn't matter
+            } else {
+                query = constantFieldType.wildcardQuery(value, caseInsensitive, context); // the rewrite method doesn't matter
+            }
             if (query instanceof MatchAllDocsQuery) {
                 return new MatchAllQueryBuilder();
             } else if (query instanceof MatchNoDocsQuery) {
@@ -244,7 +282,7 @@ public class WildcardQueryBuilder extends AbstractQueryBuilder<WildcardQueryBuil
 
     @Override
     protected int doHashCode() {
-        return Objects.hash(fieldName, value, rewrite, caseInsensitive);
+        return Objects.hash(fieldName, value, rewrite, caseInsensitive, forceStringMatch);
     }
 
     @Override
@@ -252,7 +290,8 @@ public class WildcardQueryBuilder extends AbstractQueryBuilder<WildcardQueryBuil
         return Objects.equals(fieldName, other.fieldName)
             && Objects.equals(value, other.value)
             && Objects.equals(rewrite, other.rewrite)
-            && Objects.equals(caseInsensitive, other.caseInsensitive);
+            && Objects.equals(caseInsensitive, other.caseInsensitive)
+            && Objects.equals(forceStringMatch, other.forceStringMatch);
     }
 
     @Override

server/src/test/java/org/elasticsearch/index/query/WildcardQueryBuilderTests.java

@@ -114,6 +114,7 @@ public class WildcardQueryBuilderTests extends AbstractQueryTestCase<WildcardQue
         checkGeneratedJson(json, parsed);
         assertEquals(json, "ki*y", parsed.value());
         assertEquals(json, 2.0, parsed.boost(), 0.0001);
+        assertEquals(new WildcardQueryBuilder("user", "ki*y", false).caseInsensitive(true).boost(2.0f), parsed);
     }
 
     public void testParseFailsWithMultipleFields() throws IOException {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/cluster/routing/allocation/mapper/DataTierFieldMapper.java

@@ -60,6 +60,11 @@ public class DataTierFieldMapper extends MetadataFieldMapper {
             return Regex.simpleMatch(pattern, tierPreference);
         }
 
+        @Override
+        public String getConstantFieldValue(SearchExecutionContext context) {
+            return context.getTierPreference();
+        }
+
         @Override
         public Query existsQuery(SearchExecutionContext context) {
             String tierPreference = context.getTierPreference();

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/AutomatonQuery.java

@@ -1,66 +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.core.querydsl.query;
-
-import org.apache.lucene.util.automaton.Automaton;
-import org.elasticsearch.index.query.AutomatonQueryBuilder;
-import org.elasticsearch.index.query.QueryBuilder;
-import org.elasticsearch.xpack.esql.core.tree.Source;
-
-import java.util.Objects;
-
-/**
- * Query that matches documents based on a Lucene Automaton.
- */
-public class AutomatonQuery extends Query {
-
-    private final String field;
-    private final Automaton automaton;
-    private final String automatonDescription;
-
-    public AutomatonQuery(Source source, String field, Automaton automaton, String automatonDescription) {
-        super(source);
-        this.field = field;
-        this.automaton = automaton;
-        this.automatonDescription = automatonDescription;
-    }
-
-    public String field() {
-        return field;
-    }
-
-    @Override
-    protected QueryBuilder asBuilder() {
-        return new AutomatonQueryBuilder(field, automaton, automatonDescription);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(field, automaton, automatonDescription);
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-        if (this == obj) {
-            return true;
-        }
-
-        if (obj == null || getClass() != obj.getClass()) {
-            return false;
-        }
-
-        AutomatonQuery other = (AutomatonQuery) obj;
-        return Objects.equals(field, other.field)
-            && Objects.equals(automaton, other.automaton)
-            && Objects.equals(automatonDescription, other.automatonDescription);
-    }
-
-    @Override
-    protected String innerToString() {
-        return "AutomatonQuery{" + "field='" + field + '\'' + '}';
-    }
-}

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/BoolQuery.java

@@ -100,4 +100,14 @@ public class BoolQuery extends Query {
     public boolean scorable() {
         return true;
     }
+
+    @Override
+    public boolean containsPlan() {
+        for (Query q : queries) {
+            if (q.containsPlan()) {
+                return true;
+            }
+        }
+        return false;
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/ExistsQuery.java

@@ -29,4 +29,9 @@ public class ExistsQuery extends Query {
     protected String innerToString() {
         return name;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/GeoDistanceQuery.java

@@ -75,4 +75,9 @@ public class GeoDistanceQuery extends Query {
     protected String innerToString() {
         return field + ":" + "(" + distance + "," + "(" + lat + ", " + lon + "))";
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/MatchAll.java

@@ -25,4 +25,9 @@ public class MatchAll extends Query {
     protected String innerToString() {
         return "";
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/NotQuery.java

@@ -59,4 +59,9 @@ public class NotQuery extends Query {
     public Query negate(Source source) {
         return child;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return child.containsPlan();
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/PrefixQuery.java

@@ -61,4 +61,9 @@ public class PrefixQuery extends Query {
     protected String innerToString() {
         return field + ":" + query;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/Query.java

@@ -70,6 +70,12 @@ public abstract class Query {
      */
     protected abstract String innerToString();
 
+    /**
+     * Does the result of calling {@link #asBuilder()} need the plan
+     * to serialize itself?
+     */
+    public abstract boolean containsPlan();
+
     @Override
     public boolean equals(Object obj) {
         if (obj == null || obj.getClass() != getClass()) {

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/QueryStringQuery.java

@@ -137,4 +137,9 @@ public class QueryStringQuery extends Query {
     public boolean scorable() {
         return true;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/RangeQuery.java

@@ -118,4 +118,9 @@ public class RangeQuery extends Query {
     protected String innerToString() {
         return field + ":" + (includeLower ? "[" : "(") + lower + ", " + upper + (includeUpper ? "]" : ")") + "@" + zoneId.getId();
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/RegexQuery.java

@@ -69,4 +69,9 @@ public class RegexQuery extends Query {
     protected String innerToString() {
         return field + "~ /" + regex + "/";
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/TermQuery.java

@@ -91,4 +91,9 @@ public class TermQuery extends Query {
     public boolean scorable() {
         return scorable;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/TermsQuery.java

@@ -53,4 +53,9 @@ public class TermsQuery extends Query {
     protected String innerToString() {
         return term + ":" + values;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/querydsl/query/WildcardQuery.java

@@ -12,22 +12,18 @@ import org.elasticsearch.xpack.esql.core.tree.Source;
 
 import java.util.Objects;
 
-import static org.elasticsearch.index.query.QueryBuilders.wildcardQuery;
-
 public class WildcardQuery extends Query {
 
     private final String field, query;
     private final boolean caseInsensitive;
+    private final boolean forceStringMatch;
 
-    public WildcardQuery(Source source, String field, String query) {
-        this(source, field, query, false);
-    }
-
-    public WildcardQuery(Source source, String field, String query, boolean caseInsensitive) {
+    public WildcardQuery(Source source, String field, String query, boolean caseInsensitive, boolean forceStringMatch) {
         super(source);
         this.field = field;
         this.query = query;
         this.caseInsensitive = caseInsensitive;
+        this.forceStringMatch = forceStringMatch;
     }
 
     public String field() {
@@ -44,14 +40,14 @@ public class WildcardQuery extends Query {
 
     @Override
     protected QueryBuilder asBuilder() {
-        WildcardQueryBuilder wb = wildcardQuery(field, query);
+        WildcardQueryBuilder wb = new WildcardQueryBuilder(field, query, forceStringMatch);
         // ES does not allow case_insensitive to be set to "false", it should be either "true" or not specified
         return caseInsensitive == false ? wb : wb.caseInsensitive(caseInsensitive);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(field, query, caseInsensitive);
+        return Objects.hash(field, query, caseInsensitive, forceStringMatch);
     }
 
     @Override
@@ -67,11 +63,17 @@ public class WildcardQuery extends Query {
         WildcardQuery other = (WildcardQuery) obj;
         return Objects.equals(field, other.field)
             && Objects.equals(query, other.query)
-            && Objects.equals(caseInsensitive, other.caseInsensitive);
+            && Objects.equals(caseInsensitive, other.caseInsensitive)
+            && Objects.equals(forceStringMatch, other.forceStringMatch);
     }
 
     @Override
     protected String innerToString() {
         return field + ":" + query;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/querydsl/query/LeafQueryTests.java

@@ -31,6 +31,11 @@ public class LeafQueryTests extends ESTestCase {
         protected String innerToString() {
             return "";
         }
+
+        @Override
+        public boolean containsPlan() {
+            return false;
+        }
     }
 
     public void testEqualsAndHashCode() {

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClustersIT.java

@@ -33,6 +33,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
@@ -176,11 +177,7 @@ public class MultiClustersIT extends ESRestTestCase {
             requiredCapabilities.add("like_list_on_index_fields");
         }
         // the feature is completely supported if both local and remote clusters support it
-        boolean isSupported = clusterHasCapability("POST", "/_query", List.of(), requiredCapabilities).orElse(false);
-        try (RestClient remoteClient = remoteClusterClient()) {
-            isSupported = isSupported
-                && clusterHasCapability(remoteClient, "POST", "/_query", List.of(), requiredCapabilities).orElse(false);
-        }
+        boolean isSupported = capabilitiesSupportedNewAndOld(requiredCapabilities);
 
         if (isSupported) {
             assertResultMap(includeCCSMetadata, result, columns, values, remoteOnly);
@@ -433,6 +430,53 @@ public class MultiClustersIT extends ESRestTestCase {
         assertResultMapForLike(includeCCSMetadata, result, columns, values, false, false);
     }
 
+    public void testLikeIndexLegacySettingNoResults() throws Exception {
+        // the feature is completely supported if both local and remote clusters support it
+        assumeTrue("not supported", capabilitiesSupportedNewAndOld(List.of("like_on_index_fields")));
+        try (
+            ClusterSettingToggle ignored = new ClusterSettingToggle(adminClient(), "esql.query.string_like_on_index", false, true);
+            RestClient remoteClient = remoteClusterClient();
+            ClusterSettingToggle ignored2 = new ClusterSettingToggle(remoteClient, "esql.query.string_like_on_index", false, true)
+        ) {
+            // test code with the setting changed
+            boolean includeCCSMetadata = includeCCSMetadata();
+            Map<String, Object> result = run("""
+                FROM test-local-index,*:test-remote-index METADATA _index
+                | WHERE _index LIKE "*remote*"
+                | STATS c = COUNT(*) BY _index
+                | SORT _index ASC
+                """, includeCCSMetadata);
+            var columns = List.of(Map.of("name", "c", "type", "long"), Map.of("name", "_index", "type", "keyword"));
+            // we expect empty result, since the setting is false
+            var values = List.of();
+            assertResultMapForLike(includeCCSMetadata, result, columns, values, false, false);
+        }
+    }
+
+    public void testLikeIndexLegacySettingResults() throws Exception {
+        // we require that the admin client supports the like_on_index_fields capability
+        // otherwise we will get an error when trying to toggle the setting
+        // the remote client does not have to support it
+        assumeTrue("not supported", capabilitiesSupportedNewAndOld(List.of("like_on_index_fields")));
+        try (
+            ClusterSettingToggle ignored = new ClusterSettingToggle(adminClient(), "esql.query.string_like_on_index", false, true);
+            RestClient remoteClient = remoteClusterClient();
+            ClusterSettingToggle ignored2 = new ClusterSettingToggle(remoteClient, "esql.query.string_like_on_index", false, true)
+        ) {
+            boolean includeCCSMetadata = includeCCSMetadata();
+            Map<String, Object> result = run("""
+                FROM test-local-index,*:test-remote-index METADATA _index
+                | WHERE _index LIKE "*remote*:*remote*"
+                | STATS c = COUNT(*) BY _index
+                | SORT _index ASC
+                """, includeCCSMetadata);
+            var columns = List.of(Map.of("name", "c", "type", "long"), Map.of("name", "_index", "type", "keyword"));
+            // we expect results, since the setting is false, but there is : in the LIKE query
+            var values = List.of(List.of(remoteDocs.size(), REMOTE_CLUSTER_NAME + ":" + remoteIndex));
+            assertResultMapForLike(includeCCSMetadata, result, columns, values, false, false);
+        }
+    }
+
     public void testNotLikeIndex() throws Exception {
         boolean includeCCSMetadata = includeCCSMetadata();
         Map<String, Object> result = run("""
@@ -447,12 +491,8 @@ public class MultiClustersIT extends ESRestTestCase {
     }
 
     public void testLikeListIndex() throws Exception {
-        List<String> requiredCapabilities = new ArrayList<>(List.of("like_list_on_index_fields"));
         // the feature is completely supported if both local and remote clusters support it
-        if (capabilitiesSupportedNewAndOld(requiredCapabilities) == false) {
-            logger.info("-->  skipping testNotLikeListIndex, due to missing capability");
-            return;
-        }
+        assumeTrue("not supported", capabilitiesSupportedNewAndOld(List.of("like_list_on_index_fields")));
         boolean includeCCSMetadata = includeCCSMetadata();
         Map<String, Object> result = run("""
             FROM test-local-index,*:test-remote-index METADATA _index
@@ -466,12 +506,7 @@ public class MultiClustersIT extends ESRestTestCase {
     }
 
     public void testNotLikeListIndex() throws Exception {
-        List<String> requiredCapabilities = new ArrayList<>(List.of("like_list_on_index_fields"));
-        // the feature is completely supported if both local and remote clusters support it
-        if (capabilitiesSupportedNewAndOld(requiredCapabilities) == false) {
-            logger.info("-->  skipping testNotLikeListIndex, due to missing capability");
-            return;
-        }
+        assumeTrue("not supported", capabilitiesSupportedNewAndOld(List.of("like_list_on_index_fields")));
         boolean includeCCSMetadata = includeCCSMetadata();
         Map<String, Object> result = run("""
             FROM test-local-index,*:test-remote-index METADATA _index
@@ -484,13 +519,8 @@ public class MultiClustersIT extends ESRestTestCase {
         assertResultMapForLike(includeCCSMetadata, result, columns, values, false, true);
     }
 
-    public void testNotLikeListKeyWord() throws Exception {
-        List<String> requiredCapabilities = new ArrayList<>(List.of("like_list_on_index_fields"));
-        // the feature is completely supported if both local and remote clusters support it
-        if (capabilitiesSupportedNewAndOld(requiredCapabilities) == false) {
-            logger.info("-->  skipping testNotLikeListIndex, due to missing capability");
-            return;
-        }
+    public void testNotLikeListKeyword() throws Exception {
+        assumeTrue("not supported", capabilitiesSupportedNewAndOld(List.of("like_with_list_of_patterns")));
         boolean includeCCSMetadata = includeCCSMetadata();
         Map<String, Object> result = run("""
             FROM test-local-index,*:test-remote-index METADATA _index
@@ -499,7 +529,11 @@ public class MultiClustersIT extends ESRestTestCase {
             | SORT _index ASC
             """, includeCCSMetadata);
         var columns = List.of(Map.of("name", "c", "type", "long"), Map.of("name", "_index", "type", "keyword"));
-        var values = List.of(List.of(localDocs.size(), localIndex));
+        Predicate<Doc> filter = d -> false == (d.color.contains("blue") || d.color.contains("red"));
+        var values = List.of(
+            List.of((int) remoteDocs.stream().filter(filter).count(), REMOTE_CLUSTER_NAME + ":" + remoteIndex),
+            List.of((int) localDocs.stream().filter(filter).count(), localIndex)
+        );
         assertResultMapForLike(includeCCSMetadata, result, columns, values, false, true);
     }
 
@@ -545,4 +579,28 @@ public class MultiClustersIT extends ESRestTestCase {
     private static boolean includeCCSMetadata() {
         return ccsMetadataAvailable() && randomBoolean();
     }
+
+    public static class ClusterSettingToggle implements AutoCloseable {
+        private final RestClient client;
+        private final String settingKey;
+        private final Object originalValue;
+
+        public ClusterSettingToggle(RestClient client, String settingKey, Object newValue, Object restoreValue) throws IOException {
+            this.client = client;
+            this.settingKey = settingKey;
+            this.originalValue = restoreValue;
+            setValue(newValue);
+        }
+
+        private void setValue(Object value) throws IOException {
+            Request set = new Request("PUT", "/_cluster/settings");
+            set.setJsonEntity("{\"persistent\": {\"" + settingKey + "\": " + value + "}}");
+            ESRestTestCase.assertOK(client.performRequest(set));
+        }
+
+        @Override
+        public void close() throws IOException {
+            setValue(originalValue == null ? "null" : originalValue);
+        }
+    }
 }

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

@@ -863,7 +863,7 @@ public final class EsqlTestUtils {
     }
 
     public static WildcardLike wildcardLike(Expression left, String exp) {
-        return new WildcardLike(EMPTY, left, new WildcardPattern(exp));
+        return new WildcardLike(EMPTY, left, new WildcardPattern(exp), false);
     }
 
     public static RLike rlike(Expression left, String exp) {

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

@@ -1212,6 +1212,8 @@ public class EsqlCapabilities {
 
         LIKE_WITH_LIST_OF_PATTERNS,
 
+        LIKE_LIST_ON_INDEX_FIELDS,
+
         /**
          * Support parameters for SAMPLE command.
          */
@@ -1229,7 +1231,10 @@ public class EsqlCapabilities {
          * (Re)Added EXPLAIN command
          */
         EXPLAIN(Build.current().isSnapshot()),
-
+        /**
+         * Support improved behavior for LIKE operator when used with index fields.
+         */
+        LIKE_ON_INDEX_FIELDS,
         /**
          * Support avg with aggregate metric doubles
          */

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/capabilities/TranslationAware.java

@@ -7,8 +7,11 @@
 
 package org.elasticsearch.xpack.esql.capabilities;
 
+import org.apache.lucene.search.MultiTermQuery.RewriteMethod;
 import org.elasticsearch.compute.lucene.LuceneTopNSourceOperator;
 import org.elasticsearch.compute.operator.FilterOperator;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.querydsl.query.Query;
 import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.LucenePushdownPredicates;
@@ -47,6 +50,23 @@ public interface TranslationAware {
      */
     Query asQuery(LucenePushdownPredicates pushdownPredicates, TranslatorHandler handler);
 
+    /**
+     * Translates this expression into a Lucene {@link org.apache.lucene.search.Query}.
+     * <p>
+     * Implementations should use the provided field type, rewrite method, and search execution context
+     * to construct an appropriate Lucene query for this expression.
+     * By default, this method throws {@link UnsupportedOperationException}; override it in subclasses
+     * that support Lucene query translation.
+     * </p>
+     */
+    default org.apache.lucene.search.Query asLuceneQuery(
+        MappedFieldType fieldType,
+        RewriteMethod constantScoreRewrite,
+        SearchExecutionContext context
+    ) {
+        throw new UnsupportedOperationException("asLuceneQuery is not implemented for " + getClass().getName());
+    }
+
     /**
      * Subinterface for expressions that can only process single values (and null out on MVs).
      */

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

@@ -151,7 +151,7 @@ public class EndsWith extends EsqlScalarFunction implements TranslationAware.Sin
         // TODO: Get the real FoldContext here
         var wildcardQuery = "*" + QueryParser.escape(BytesRefs.toString(suffix.fold(FoldContext.small())));
 
-        return new WildcardQuery(source(), fieldName, wildcardQuery);
+        return new WildcardQuery(source(), fieldName, wildcardQuery, false, false);
     }
 
     @Override

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

@@ -148,7 +148,7 @@ public class StartsWith extends EsqlScalarFunction implements TranslationAware.S
         // TODO: Get the real FoldContext here
         var wildcardQuery = QueryParser.escape(BytesRefs.toString(prefix.fold(FoldContext.small()))) + "*";
 
-        return new WildcardQuery(source(), fieldName, wildcardQuery);
+        return new WildcardQuery(source(), fieldName, wildcardQuery, false, false);
     }
 
     @Override

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/regex/WildcardLike.java

@@ -122,11 +122,14 @@ public class WildcardLike extends RegexMatch<WildcardPattern> {
     public Query asQuery(LucenePushdownPredicates pushdownPredicates, TranslatorHandler handler) {
         var field = field();
         LucenePushdownPredicates.checkIsPushableAttribute(field);
-        return translateField(handler.nameOf(field instanceof FieldAttribute fa ? fa.exactAttribute() : field));
+        return translateField(
+            handler.nameOf(field instanceof FieldAttribute fa ? fa.exactAttribute() : field),
+            pushdownPredicates.flags().stringLikeOnIndex()
+        );
     }
 
     // TODO: see whether escaping is needed
-    private Query translateField(String targetFieldName) {
-        return new WildcardQuery(source(), targetFieldName, pattern().asLuceneWildcard(), caseInsensitive());
+    private Query translateField(String targetFieldName, boolean forceStringMatch) {
+        return new WildcardQuery(source(), targetFieldName, pattern().asLuceneWildcard(), caseInsensitive(), forceStringMatch);
     }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/regex/WildcardLikeList.java

@@ -7,26 +7,35 @@
 
 package org.elasticsearch.xpack.esql.expression.function.scalar.string.regex;
 
+import org.apache.lucene.search.MultiTermQuery.RewriteMethod;
+import org.apache.lucene.util.automaton.Automaton;
+import org.apache.lucene.util.automaton.CharacterRunAutomaton;
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
 import org.elasticsearch.xpack.esql.core.expression.predicate.regex.WildcardPattern;
 import org.elasticsearch.xpack.esql.core.expression.predicate.regex.WildcardPatternList;
-import org.elasticsearch.xpack.esql.core.querydsl.query.AutomatonQuery;
 import org.elasticsearch.xpack.esql.core.querydsl.query.Query;
 import org.elasticsearch.xpack.esql.core.querydsl.query.WildcardQuery;
 import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.expression.function.Param;
+import org.elasticsearch.xpack.esql.io.stream.ExpressionQuery;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.LucenePushdownPredicates;
 import org.elasticsearch.xpack.esql.planner.TranslatorHandler;
 
 import java.io.IOException;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
+import static org.elasticsearch.index.query.WildcardQueryBuilder.expressionTransportSupported;
+
 public class WildcardLikeList extends RegexMatch<WildcardPatternList> {
     public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
         Expression.class,
@@ -34,6 +43,30 @@ public class WildcardLikeList extends RegexMatch<WildcardPatternList> {
         WildcardLikeList::new
     );
 
+    Supplier<Automaton> automatonSupplier = new Supplier<>() {
+        Automaton cached;
+
+        @Override
+        public Automaton get() {
+            if (cached == null) {
+                cached = pattern().createAutomaton(caseInsensitive());
+            }
+            return cached;
+        }
+    };
+
+    Supplier<CharacterRunAutomaton> characterRunAutomatonSupplier = new Supplier<>() {
+        CharacterRunAutomaton cached;
+
+        @Override
+        public CharacterRunAutomaton get() {
+            if (cached == null) {
+                cached = new CharacterRunAutomaton(automatonSupplier.get());
+            }
+            return cached;
+        }
+    };
+
     /**
      * The documentation for this function is in WildcardLike, and shown to the users `LIKE` in the docs.
      */
@@ -92,10 +125,10 @@ public class WildcardLikeList extends RegexMatch<WildcardPatternList> {
      */
     @Override
     public Translatable translatable(LucenePushdownPredicates pushdownPredicates) {
-        if (pushdownPredicates.minTransportVersion() == null) {
+        if (supportsPushdown(pushdownPredicates.minTransportVersion())) {
             return pushdownPredicates.isPushableAttribute(field()) ? Translatable.YES : Translatable.NO;
         } else {
-            // The AutomatonQuery that we use right now isn't serializable.
+            // The ExpressionQuery we use isn't serializable to all nodes in the cluster.
             return Translatable.NO;
         }
     }
@@ -108,20 +141,40 @@ public class WildcardLikeList extends RegexMatch<WildcardPatternList> {
     public Query asQuery(LucenePushdownPredicates pushdownPredicates, TranslatorHandler handler) {
         var field = field();
         LucenePushdownPredicates.checkIsPushableAttribute(field);
-        return translateField(handler.nameOf(field instanceof FieldAttribute fa ? fa.exactAttribute() : field));
+        String targetFieldName = handler.nameOf(field instanceof FieldAttribute fa ? fa.exactAttribute() : field);
+        return translateField(targetFieldName);
     }
 
-    /**
-     * Translates the field to a {@link WildcardQuery} using the first pattern in the list.
-     * Throws an {@link IllegalArgumentException} if the pattern list contains more than one pattern.
-     */
-    private Query translateField(String targetFieldName) {
-        return new AutomatonQuery(source(), targetFieldName, pattern().createAutomaton(caseInsensitive()), getAutomatonDescription());
+    private boolean supportsPushdown(TransportVersion version) {
+        return version == null || expressionTransportSupported(version);
     }
 
-    private String getAutomatonDescription() {
+    @Override
+    public org.apache.lucene.search.Query asLuceneQuery(
+        MappedFieldType fieldType,
+        RewriteMethod constantScoreRewrite,
+        SearchExecutionContext context
+    ) {
+        return fieldType.automatonQuery(
+            automatonSupplier,
+            characterRunAutomatonSupplier,
+            constantScoreRewrite,
+            context,
+            getLuceneQueryDescription()
+        );
+    }
+
+    private String getLuceneQueryDescription() {
         // we use the information used to create the automaton to describe the query here
         String patternDesc = pattern().patternList().stream().map(WildcardPattern::pattern).collect(Collectors.joining("\", \""));
         return "LIKE(\"" + patternDesc + "\"), caseInsensitive=" + caseInsensitive();
     }
+
+    /**
+     * Translates the field to a {@link WildcardQuery} using the first pattern in the list.
+     * Throws an {@link IllegalArgumentException} if the pattern list contains more than one pattern.
+     */
+    private Query translateField(String targetFieldName) {
+        return new ExpressionQuery(source(), targetFieldName, this);
+    }
 }

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

@@ -0,0 +1,67 @@
+/*
+ * 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.io.stream;
+
+import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.querydsl.query.Query;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+
+import java.util.Objects;
+
+/**
+ * Implements an Expression query, which matches documents based on a given expression.
+ */
+public class ExpressionQuery extends Query {
+
+    private final String targetFieldName;
+    private final Expression expression;
+
+    public ExpressionQuery(Source source, String targetFieldName, Expression expression) {
+        super(source);
+        this.targetFieldName = targetFieldName;
+        this.expression = expression;
+    }
+
+    public String field() {
+        return targetFieldName;
+    }
+
+    @Override
+    protected QueryBuilder asBuilder() {
+        return new ExpressionQueryBuilder(targetFieldName, expression);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(targetFieldName, expression);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+
+        ExpressionQuery other = (ExpressionQuery) obj;
+        return Objects.equals(targetFieldName, other.targetFieldName) && Objects.equals(expression, other.expression);
+    }
+
+    @Override
+    protected String innerToString() {
+        return "ExpressionQuery{" + "field='" + targetFieldName + '\'' + '}';
+    }
+
+    @Override
+    public boolean containsPlan() {
+        return true;
+    }
+}

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

@@ -0,0 +1,123 @@
+/*
+ * 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.io.stream;
+
+import org.apache.lucene.search.MatchNoDocsQuery;
+import org.apache.lucene.search.Query;
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.query.AbstractQueryBuilder;
+import org.elasticsearch.index.query.MultiTermQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.index.query.SearchExecutionContext;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.esql.capabilities.TranslationAware;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import static org.apache.lucene.search.MultiTermQuery.CONSTANT_SCORE_REWRITE;
+
+/**
+ * Implements an Expression query builder, which matches documents based on a given expression.
+ * The expression itself must provide the {@link TranslationAware#asLuceneQuery} interface to be translated into a Lucene query.
+ * It allows for serialization of the expression and generate an AutomatonQuery on the data node
+ * as Automaton does not support serialization.
+ */
+public class ExpressionQueryBuilder extends AbstractQueryBuilder<ExpressionQueryBuilder> implements MultiTermQueryBuilder {
+    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
+        QueryBuilder.class,
+        "expressionQueryBuilder",
+        ExpressionQueryBuilder::new
+    );
+    private final String fieldName;
+    private final Expression expression;
+
+    public ExpressionQueryBuilder(String fieldName, Expression expression) {
+        if (Strings.isEmpty(fieldName)) {
+            throw new IllegalArgumentException("field name is null or empty");
+        }
+        if (expression == null) {
+            throw new IllegalArgumentException("expression cannot be null");
+        }
+        this.fieldName = fieldName;
+        this.expression = expression;
+    }
+
+    /**
+     * Read from a stream.
+     */
+    private ExpressionQueryBuilder(StreamInput in) throws IOException {
+        super(in);
+        fieldName = in.readString();
+        assert in instanceof PlanStreamInput;
+        this.expression = in.readNamedWriteable(Expression.class);
+    }
+
+    public Expression getExpression() {
+        return expression;
+    }
+
+    @Override
+    protected void doWriteTo(StreamOutput out) throws IOException {
+        out.writeString(this.fieldName);
+        assert out instanceof PlanStreamOutput;
+        out.writeNamedWriteable(expression);
+    }
+
+    @Override
+    public String fieldName() {
+        return fieldName;
+    }
+
+    @Override
+    public String getWriteableName() {
+        return ENTRY.name;
+    }
+
+    @Override
+    protected void doXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject(ENTRY.name); // Use the appropriate query name
+        builder.field("field", fieldName);
+        builder.field("expression", expression.toString());
+        builder.endObject();
+    }
+
+    @Override
+    protected Query doToQuery(SearchExecutionContext context) {
+        if (expression instanceof TranslationAware translationAware) {
+            MappedFieldType fieldType = context.getFieldType(fieldName);
+            if (fieldType == null) {
+                return new MatchNoDocsQuery("Field [" + fieldName + "] does not exist");
+            }
+            return translationAware.asLuceneQuery(fieldType, CONSTANT_SCORE_REWRITE, context);
+        } else {
+            throw new UnsupportedOperationException("ExpressionQueryBuilder does not support non-automaton expressions");
+        }
+    }
+
+    @Override
+    protected int doHashCode() {
+        return Objects.hash(fieldName, expression);
+    }
+
+    @Override
+    protected boolean doEquals(ExpressionQueryBuilder other) {
+        return Objects.equals(fieldName, other.fieldName) && Objects.equals(expression, other.expression);
+    }
+
+    @Override
+    public TransportVersion getMinimalSupportedVersion() {
+        throw new UnsupportedOperationException("AutomatonQueryBuilder does not support getMinimalSupportedVersion");
+    }
+}

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

@@ -0,0 +1,111 @@
+/*
+ * 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.io.stream;
+
+import org.apache.lucene.search.Query;
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.TransportVersions;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.index.query.SearchExecutionContext;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.esql.session.Configuration;
+
+import java.io.IOException;
+
+import static org.elasticsearch.index.query.WildcardQueryBuilder.expressionTransportSupported;
+
+/**
+ * A {@link QueryBuilder} that wraps another {@linkplain QueryBuilder}
+ * so it read with a {@link PlanStreamInput}.
+ */
+public class PlanStreamWrapperQueryBuilder implements QueryBuilder {
+    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
+        QueryBuilder.class,
+        "planwrapper",
+        PlanStreamWrapperQueryBuilder::new
+    );
+
+    private final Configuration configuration;
+    private final QueryBuilder next;
+
+    public PlanStreamWrapperQueryBuilder(Configuration configuration, QueryBuilder next) {
+        this.configuration = configuration;
+        this.next = next;
+    }
+
+    public PlanStreamWrapperQueryBuilder(StreamInput in) throws IOException {
+        configuration = Configuration.readWithoutTables(in);
+        PlanStreamInput planStreamInput = new PlanStreamInput(in, in.namedWriteableRegistry(), configuration);
+        next = planStreamInput.readNamedWriteable(QueryBuilder.class);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        configuration.withoutTables().writeTo(out);
+        new PlanStreamOutput(out, configuration).writeNamedWriteable(next);
+    }
+
+    @Override
+    public TransportVersion getMinimalSupportedVersion() {
+        return TransportVersions.ESQL_FIXED_INDEX_LIKE_9_1;
+    }
+
+    @Override
+    public boolean supportsVersion(TransportVersion version) {
+        return expressionTransportSupported(version);
+    }
+
+    @Override
+    public Query toQuery(SearchExecutionContext context) throws IOException {
+        return next.toQuery(context);
+    }
+
+    @Override
+    public QueryBuilder queryName(String queryName) {
+        next.queryName(queryName);
+        return this;
+    }
+
+    @Override
+    public String queryName() {
+        return next.queryName();
+    }
+
+    @Override
+    public float boost() {
+        return next.boost();
+    }
+
+    @Override
+    public QueryBuilder boost(float boost) {
+        next.boost(boost);
+        return this;
+    }
+
+    @Override
+    public String getName() {
+        return getWriteableName();
+    }
+
+    @Override
+    public String getWriteableName() {
+        return ENTRY.name;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        return next.toXContent(builder, params);
+    }
+
+    public QueryBuilder next() {
+        return next;
+    }
+}

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

@@ -8,7 +8,8 @@
 package org.elasticsearch.xpack.esql.optimizer;
 
 import org.elasticsearch.xpack.esql.core.expression.FoldContext;
+import org.elasticsearch.xpack.esql.plugin.EsqlFlags;
 import org.elasticsearch.xpack.esql.session.Configuration;
 import org.elasticsearch.xpack.esql.stats.SearchStats;
 
-public record LocalPhysicalOptimizerContext(Configuration configuration, FoldContext foldCtx, SearchStats searchStats) {}
+public record LocalPhysicalOptimizerContext(EsqlFlags flags, Configuration configuration, FoldContext foldCtx, SearchStats searchStats) {}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/physical/local/EnableSpatialDistancePushdown.java

@@ -77,14 +77,14 @@ public class EnableSpatialDistancePushdown extends PhysicalOptimizerRules.Parame
     protected PhysicalPlan rule(FilterExec filterExec, LocalPhysicalOptimizerContext ctx) {
         PhysicalPlan plan = filterExec;
         if (filterExec.child() instanceof EsQueryExec esQueryExec) {
-            plan = rewrite(ctx.foldCtx(), filterExec, esQueryExec, LucenePushdownPredicates.from(ctx.searchStats()));
+            plan = rewrite(ctx.foldCtx(), filterExec, esQueryExec, LucenePushdownPredicates.from(ctx.searchStats(), ctx.flags()));
         } else if (filterExec.child() instanceof EvalExec evalExec && evalExec.child() instanceof EsQueryExec esQueryExec) {
             plan = rewriteBySplittingFilter(
                 ctx.foldCtx(),
                 filterExec,
                 evalExec,
                 esQueryExec,
-                LucenePushdownPredicates.from(ctx.searchStats())
+                LucenePushdownPredicates.from(ctx.searchStats(), ctx.flags())
             );
         }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/physical/local/LucenePushdownPredicates.java

@@ -16,6 +16,7 @@ import org.elasticsearch.xpack.esql.core.expression.MetadataAttribute;
 import org.elasticsearch.xpack.esql.core.expression.TypedAttribute;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.util.Check;
+import org.elasticsearch.xpack.esql.plugin.EsqlFlags;
 import org.elasticsearch.xpack.esql.stats.SearchStats;
 
 /**
@@ -51,6 +52,8 @@ public interface LucenePushdownPredicates {
     @Nullable
     TransportVersion minTransportVersion();
 
+    EsqlFlags flags();
+
     /**
      * For TEXT fields, we need to check if the field has a subfield of type KEYWORD that can be used instead.
      */
@@ -122,18 +125,23 @@ public interface LucenePushdownPredicates {
      * In particular, it assumes TEXT fields have no exact subfields (underlying keyword field),
      * and that isAggregatable means indexed and has hasDocValues.
      */
-    LucenePushdownPredicates DEFAULT = forCanMatch(null);
+    LucenePushdownPredicates DEFAULT = forCanMatch(null, new EsqlFlags(true));
 
     /**
      * A {@link LucenePushdownPredicates} for use with the {@code can_match} phase.
      */
-    static LucenePushdownPredicates forCanMatch(TransportVersion minTransportVersion) {
+    static LucenePushdownPredicates forCanMatch(TransportVersion minTransportVersion, EsqlFlags flags) {
         return new LucenePushdownPredicates() {
             @Override
             public TransportVersion minTransportVersion() {
                 return minTransportVersion;
             }
 
+            @Override
+            public EsqlFlags flags() {
+                return flags;
+            }
+
             @Override
             public boolean hasExactSubfield(FieldAttribute attr) {
                 return false;
@@ -162,7 +170,7 @@ public interface LucenePushdownPredicates {
      * If we have access to {@link SearchStats} over a collection of shards, we can make more fine-grained decisions about what can be
      * pushed down. This should open up more opportunities for lucene pushdown.
      */
-    static LucenePushdownPredicates from(SearchStats stats) {
+    static LucenePushdownPredicates from(SearchStats stats, EsqlFlags flags) {
         // TODO: use FieldAttribute#fieldName, otherwise this doesn't apply to field attributes used for union types.
         // C.f. https://github.com/elastic/elasticsearch/issues/128905
         return new LucenePushdownPredicates() {
@@ -171,6 +179,11 @@ public interface LucenePushdownPredicates {
                 return null;
             }
 
+            @Override
+            public EsqlFlags flags() {
+                return flags;
+            }
+
             @Override
             public boolean hasExactSubfield(FieldAttribute attr) {
                 return stats.hasExactSubfield(new FieldAttribute.FieldName(attr.name()));

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/physical/local/PushFiltersToSource.java

@@ -53,7 +53,7 @@ public class PushFiltersToSource extends PhysicalOptimizerRules.ParameterizedOpt
     }
 
     private static PhysicalPlan planFilterExec(FilterExec filterExec, EsQueryExec queryExec, LocalPhysicalOptimizerContext ctx) {
-        LucenePushdownPredicates pushdownPredicates = LucenePushdownPredicates.from(ctx.searchStats());
+        LucenePushdownPredicates pushdownPredicates = LucenePushdownPredicates.from(ctx.searchStats(), ctx.flags());
         List<Expression> pushable = new ArrayList<>();
         List<Expression> nonPushable = new ArrayList<>();
         for (Expression exp : splitAnd(filterExec.condition())) {
@@ -75,7 +75,7 @@ public class PushFiltersToSource extends PhysicalOptimizerRules.ParameterizedOpt
         EsQueryExec queryExec,
         LocalPhysicalOptimizerContext ctx
     ) {
-        LucenePushdownPredicates pushdownPredicates = LucenePushdownPredicates.from(ctx.searchStats());
+        LucenePushdownPredicates pushdownPredicates = LucenePushdownPredicates.from(ctx.searchStats(), ctx.flags());
         AttributeMap<Attribute> aliasReplacedBy = getAliasReplacedBy(evalExec);
         List<Expression> pushable = new ArrayList<>();
         List<Expression> nonPushable = new ArrayList<>();

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/physical/local/PushTopNToSource.java

@@ -63,7 +63,7 @@ public class PushTopNToSource extends PhysicalOptimizerRules.ParameterizedOptimi
 
     @Override
     protected PhysicalPlan rule(TopNExec topNExec, LocalPhysicalOptimizerContext ctx) {
-        Pushable pushable = evaluatePushable(ctx.foldCtx(), topNExec, LucenePushdownPredicates.from(ctx.searchStats()));
+        Pushable pushable = evaluatePushable(ctx.foldCtx(), topNExec, LucenePushdownPredicates.from(ctx.searchStats(), ctx.flags()));
         return pushable.rewrite(topNExec);
     }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/EsqlParser.java

@@ -21,6 +21,7 @@ import org.elasticsearch.logging.Logger;
 import org.elasticsearch.xpack.esql.core.util.StringUtils;
 import org.elasticsearch.xpack.esql.expression.function.EsqlFunctionRegistry;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
+import org.elasticsearch.xpack.esql.session.Configuration;
 import org.elasticsearch.xpack.esql.telemetry.PlanTelemetry;
 
 import java.util.BitSet;
@@ -98,20 +99,20 @@ public class EsqlParser {
     }
 
     // testing utility
-    public LogicalPlan createStatement(String query) {
-        return createStatement(query, new QueryParams());
+    public LogicalPlan createStatement(String query, Configuration configuration) {
+        return createStatement(query, new QueryParams(), configuration);
     }
 
     // testing utility
-    public LogicalPlan createStatement(String query, QueryParams params) {
-        return createStatement(query, params, new PlanTelemetry(new EsqlFunctionRegistry()));
+    public LogicalPlan createStatement(String query, QueryParams params, Configuration configuration) {
+        return createStatement(query, params, new PlanTelemetry(new EsqlFunctionRegistry()), configuration);
     }
 
-    public LogicalPlan createStatement(String query, QueryParams params, PlanTelemetry metrics) {
+    public LogicalPlan createStatement(String query, QueryParams params, PlanTelemetry metrics, Configuration configuration) {
         if (log.isDebugEnabled()) {
             log.debug("Parsing as statement: {}", query);
         }
-        return invokeParser(query, params, metrics, EsqlBaseParser::singleStatement, AstBuilder::plan);
+        return invokeParser(query, params, metrics, EsqlBaseParser::singleStatement, AstBuilder::plan, configuration);
     }
 
     private <T> T invokeParser(
@@ -119,7 +120,8 @@ public class EsqlParser {
         QueryParams params,
         PlanTelemetry metrics,
         Function<EsqlBaseParser, ParserRuleContext> parseFunction,
-        BiFunction<AstBuilder, ParserRuleContext, T> result
+        BiFunction<AstBuilder, ParserRuleContext, T> result,
+        Configuration configuration
     ) {
         if (query.length() > MAX_LENGTH) {
             throw new ParsingException("ESQL statement is too large [{} characters > {}]", query.length(), MAX_LENGTH);

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

@@ -26,11 +26,13 @@ import org.elasticsearch.xpack.esql.core.expression.AttributeSet;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.FoldContext;
 import org.elasticsearch.xpack.esql.core.expression.MetadataAttribute;
+import org.elasticsearch.xpack.esql.core.querydsl.query.Query;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.util.Holder;
 import org.elasticsearch.xpack.esql.core.util.Queries;
 import org.elasticsearch.xpack.esql.expression.predicate.Predicates;
+import org.elasticsearch.xpack.esql.io.stream.PlanStreamWrapperQueryBuilder;
 import org.elasticsearch.xpack.esql.optimizer.LocalLogicalOptimizerContext;
 import org.elasticsearch.xpack.esql.optimizer.LocalLogicalPlanOptimizer;
 import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext;
@@ -50,6 +52,7 @@ import org.elasticsearch.xpack.esql.plan.physical.MergeExec;
 import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
 import org.elasticsearch.xpack.esql.planner.mapper.LocalMapper;
 import org.elasticsearch.xpack.esql.planner.mapper.Mapper;
+import org.elasticsearch.xpack.esql.plugin.EsqlFlags;
 import org.elasticsearch.xpack.esql.session.Configuration;
 import org.elasticsearch.xpack.esql.stats.SearchContextStats;
 import org.elasticsearch.xpack.esql.stats.SearchStats;
@@ -163,17 +166,26 @@ public class PlannerUtils {
     }
 
     public static PhysicalPlan localPlan(
+        EsqlFlags flags,
         List<SearchExecutionContext> searchContexts,
         Configuration configuration,
         FoldContext foldCtx,
         PhysicalPlan plan
     ) {
-        return localPlan(configuration, foldCtx, plan, SearchContextStats.from(searchContexts));
+        return localPlan(flags, configuration, foldCtx, plan, SearchContextStats.from(searchContexts));
     }
 
-    public static PhysicalPlan localPlan(Configuration configuration, FoldContext foldCtx, PhysicalPlan plan, SearchStats searchStats) {
+    public static PhysicalPlan localPlan(
+        EsqlFlags flags,
+        Configuration configuration,
+        FoldContext foldCtx,
+        PhysicalPlan plan,
+        SearchStats searchStats
+    ) {
         final var logicalOptimizer = new LocalLogicalPlanOptimizer(new LocalLogicalOptimizerContext(configuration, foldCtx, searchStats));
-        var physicalOptimizer = new LocalPhysicalPlanOptimizer(new LocalPhysicalOptimizerContext(configuration, foldCtx, searchStats));
+        var physicalOptimizer = new LocalPhysicalPlanOptimizer(
+            new LocalPhysicalOptimizerContext(flags, configuration, foldCtx, searchStats)
+        );
 
         return localPlan(plan, logicalOptimizer, physicalOptimizer);
     }
@@ -213,8 +225,13 @@ public class PlannerUtils {
     /**
      * Extracts a filter that can be used to skip unmatched shards on the coordinator.
      */
-    public static QueryBuilder canMatchFilter(TransportVersion minTransportVersion, PhysicalPlan plan) {
-        return detectFilter(minTransportVersion, plan, CoordinatorRewriteContext.SUPPORTED_FIELDS::contains);
+    public static QueryBuilder canMatchFilter(
+        EsqlFlags flags,
+        Configuration configuration,
+        TransportVersion minTransportVersion,
+        PhysicalPlan plan
+    ) {
+        return detectFilter(flags, configuration, minTransportVersion, plan, CoordinatorRewriteContext.SUPPORTED_FIELDS::contains);
     }
 
     /**
@@ -222,10 +239,16 @@ public class PlannerUtils {
      * We currently only use this filter for the @timestamp field, which is always a date field. Any tests that wish to use this should
      * take care to not use it with TEXT fields.
      */
-    static QueryBuilder detectFilter(TransportVersion minTransportVersion, PhysicalPlan plan, Predicate<String> fieldName) {
+    static QueryBuilder detectFilter(
+        EsqlFlags flags,
+        Configuration configuration,
+        TransportVersion minTransportVersion,
+        PhysicalPlan plan,
+        Predicate<String> fieldName
+    ) {
         // first position is the REST filter, the second the query filter
         final List<QueryBuilder> requestFilters = new ArrayList<>();
-        final LucenePushdownPredicates ctx = LucenePushdownPredicates.forCanMatch(minTransportVersion);
+        final LucenePushdownPredicates ctx = LucenePushdownPredicates.forCanMatch(minTransportVersion, flags);
         plan.forEachDown(FragmentExec.class, fe -> {
             if (fe.esFilter() != null && fe.esFilter().supportsVersion(minTransportVersion)) {
                 requestFilters.add(fe.esFilter());
@@ -253,7 +276,12 @@ public class PlannerUtils {
                     }
                 }
                 if (matches.isEmpty() == false) {
-                    requestFilters.add(TRANSLATOR_HANDLER.asQuery(ctx, Predicates.combineAnd(matches)).toQueryBuilder());
+                    Query qlQuery = TRANSLATOR_HANDLER.asQuery(ctx, Predicates.combineAnd(matches));
+                    QueryBuilder builder = qlQuery.toQueryBuilder();
+                    if (qlQuery.containsPlan()) {
+                        builder = new PlanStreamWrapperQueryBuilder(configuration, builder);
+                    }
+                    requestFilters.add(builder);
                 }
             });
         });

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

@@ -250,6 +250,7 @@ final class ClusterComputeHandler implements TransportRequestHandler<ClusterComp
         final String localSessionId = clusterAlias + ":" + globalSessionId;
         final PhysicalPlan coordinatorPlan = ComputeService.reductionPlan(plan, true);
         final AtomicReference<ComputeResponse> finalResponse = new AtomicReference<>();
+        final EsqlFlags flags = computeService.createFlags();
         final long startTimeInNanos = System.nanoTime();
         final Runnable cancelQueryOnFailure = computeService.cancelQueryOnFailure(parentTask);
         try (var computeListener = new ComputeListener(transportService.getThreadPool(), cancelQueryOnFailure, listener.map(profiles -> {
@@ -269,6 +270,7 @@ final class ClusterComputeHandler implements TransportRequestHandler<ClusterComp
                         localSessionId,
                         "remote_reduce",
                         clusterAlias,
+                        flags,
                         List.of(),
                         configuration,
                         configuration.newFoldContext(),
@@ -282,6 +284,7 @@ final class ClusterComputeHandler implements TransportRequestHandler<ClusterComp
                     localSessionId,
                     clusterAlias,
                     parentTask,
+                    flags,
                     configuration,
                     plan,
                     concreteIndices,

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

@@ -21,6 +21,7 @@ record ComputeContext(
     String sessionId,
     String description,
     String clusterAlias,
+    EsqlFlags flags,
     List<SearchContext> searchContexts,
     Configuration configuration,
     FoldContext foldCtx,

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

@@ -183,6 +183,7 @@ public class ComputeService {
     public void execute(
         String sessionId,
         CancellableTask rootTask,
+        EsqlFlags flags,
         PhysicalPlan physicalPlan,
         Configuration configuration,
         FoldContext foldContext,
@@ -195,7 +196,7 @@ public class ComputeService {
 
         // we have no sub plans, so we can just execute the given plan
         if (subplans == null || subplans.isEmpty()) {
-            executePlan(sessionId, rootTask, physicalPlan, configuration, foldContext, execInfo, null, listener, null);
+            executePlan(sessionId, rootTask, flags, physicalPlan, configuration, foldContext, execInfo, null, listener, null);
             return;
         }
 
@@ -222,6 +223,7 @@ public class ComputeService {
                 mainSessionId,
                 "main.final",
                 LOCAL_CLUSTER,
+                flags,
                 List.of(),
                 configuration,
                 foldContext,
@@ -254,6 +256,7 @@ public class ComputeService {
                     executePlan(
                         childSessionId,
                         rootTask,
+                        flags,
                         subplan,
                         configuration,
                         foldContext,
@@ -278,6 +281,7 @@ public class ComputeService {
     public void executePlan(
         String sessionId,
         CancellableTask rootTask,
+        EsqlFlags flags,
         PhysicalPlan physicalPlan,
         Configuration configuration,
         FoldContext foldContext,
@@ -322,6 +326,7 @@ public class ComputeService {
                 newChildSession(sessionId),
                 profileDescription(profileQualifier, "single"),
                 LOCAL_CLUSTER,
+                flags,
                 List.of(),
                 configuration,
                 foldContext,
@@ -408,6 +413,7 @@ public class ComputeService {
                             sessionId,
                             profileDescription(profileQualifier, "final"),
                             LOCAL_CLUSTER,
+                            flags,
                             List.of(),
                             configuration,
                             foldContext,
@@ -424,6 +430,7 @@ public class ComputeService {
                             sessionId,
                             LOCAL_CLUSTER,
                             rootTask,
+                            flags,
                             configuration,
                             dataNodePlan,
                             Set.of(localConcreteIndices.indices()),
@@ -576,7 +583,13 @@ public class ComputeService {
 
             LOGGER.debug("Received physical plan:\n{}", plan);
 
-            var localPlan = PlannerUtils.localPlan(context.searchExecutionContexts(), context.configuration(), context.foldCtx(), plan);
+            var localPlan = PlannerUtils.localPlan(
+                context.flags(),
+                context.searchExecutionContexts(),
+                context.configuration(),
+                context.foldCtx(),
+                plan
+            );
             // the planner will also set the driver parallelism in LocalExecutionPlanner.LocalExecutionPlan (used down below)
             // it's doing this in the planning of EsQueryExec (the source of the data)
             // see also EsPhysicalOperationProviders.sourcePhysicalOperation
@@ -665,6 +678,10 @@ public class ComputeService {
         }
     }
 
+    public EsqlFlags createFlags() {
+        return new EsqlFlags(clusterService.getClusterSettings());
+    }
+
     private static class ComputeGroupTaskRequest extends AbstractTransportRequest {
         private final Supplier<String> parentDescription;
 

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

@@ -99,6 +99,7 @@ final class DataNodeComputeHandler implements TransportRequestHandler<DataNodeRe
         String sessionId,
         String clusterAlias,
         CancellableTask parentTask,
+        EsqlFlags flags,
         Configuration configuration,
         PhysicalPlan dataNodePlan,
         Set<String> concreteIndices,
@@ -116,7 +117,7 @@ final class DataNodeComputeHandler implements TransportRequestHandler<DataNodeRe
             esqlExecutor,
             parentTask,
             originalIndices,
-            PlannerUtils.canMatchFilter(clusterService.state().getMinTransportVersion(), dataNodePlan),
+            PlannerUtils.canMatchFilter(flags, configuration, clusterService.state().getMinTransportVersion(), dataNodePlan),
             clusterAlias,
             configuration.allowPartialResults(),
             maxConcurrentNodesPerCluster == null ? -1 : maxConcurrentNodesPerCluster,
@@ -219,6 +220,7 @@ final class DataNodeComputeHandler implements TransportRequestHandler<DataNodeRe
     }
 
     private class DataNodeRequestExecutor {
+        private final EsqlFlags flags;
         private final DataNodeRequest request;
         private final CancellableTask parentTask;
         private final ExchangeSinkHandler exchangeSink;
@@ -229,6 +231,7 @@ final class DataNodeComputeHandler implements TransportRequestHandler<DataNodeRe
         private final Map<ShardId, Exception> shardLevelFailures;
 
         DataNodeRequestExecutor(
+            EsqlFlags flags,
             DataNodeRequest request,
             CancellableTask parentTask,
             ExchangeSinkHandler exchangeSink,
@@ -237,6 +240,7 @@ final class DataNodeComputeHandler implements TransportRequestHandler<DataNodeRe
             Map<ShardId, Exception> shardLevelFailures,
             ComputeListener computeListener
         ) {
+            this.flags = flags;
             this.request = request;
             this.parentTask = parentTask;
             this.exchangeSink = exchangeSink;
@@ -297,6 +301,7 @@ final class DataNodeComputeHandler implements TransportRequestHandler<DataNodeRe
                     sessionId,
                     "data",
                     clusterAlias,
+                    flags,
                     searchContexts,
                     configuration,
                     configuration.newFoldContext(),
@@ -422,7 +427,9 @@ final class DataNodeComputeHandler implements TransportRequestHandler<DataNodeRe
                     exchangeService.finishSinkHandler(externalId, new TaskCancelledException(task.getReasonCancelled()));
                     exchangeService.finishSinkHandler(request.sessionId(), new TaskCancelledException(task.getReasonCancelled()));
                 });
+                EsqlFlags flags = computeService.createFlags();
                 DataNodeRequestExecutor dataNodeRequestExecutor = new DataNodeRequestExecutor(
+                    flags,
                     request,
                     task,
                     internalSink,
@@ -442,6 +449,7 @@ final class DataNodeComputeHandler implements TransportRequestHandler<DataNodeRe
                         request.sessionId(),
                         "node_reduce",
                         request.clusterAlias(),
+                        flags,
                         List.of(),
                         request.configuration(),
                         new FoldContext(request.pragmas().foldLimit().getBytes()),

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

@@ -0,0 +1,34 @@
+/*
+ * 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.plugin;
+
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.settings.Setting;
+
+public class EsqlFlags {
+    public static final Setting<Boolean> ESQL_STRING_LIKE_ON_INDEX = Setting.boolSetting(
+        "esql.query.string_like_on_index",
+        true,
+        Setting.Property.NodeScope,
+        Setting.Property.Dynamic
+    );
+
+    private final boolean stringLikeOnIndex;
+
+    public EsqlFlags(boolean stringLikeOnIndex) {
+        this.stringLikeOnIndex = stringLikeOnIndex;
+    }
+
+    public EsqlFlags(ClusterSettings settings) {
+        this.stringLikeOnIndex = settings.get(ESQL_STRING_LIKE_ON_INDEX);
+    }
+
+    public boolean stringLikeOnIndex() {
+        return stringLikeOnIndex;
+    }
+}

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

@@ -72,6 +72,8 @@ import org.elasticsearch.xpack.esql.enrich.EnrichLookupOperator;
 import org.elasticsearch.xpack.esql.enrich.LookupFromIndexOperator;
 import org.elasticsearch.xpack.esql.execution.PlanExecutor;
 import org.elasticsearch.xpack.esql.expression.ExpressionWritables;
+import org.elasticsearch.xpack.esql.io.stream.ExpressionQueryBuilder;
+import org.elasticsearch.xpack.esql.io.stream.PlanStreamWrapperQueryBuilder;
 import org.elasticsearch.xpack.esql.plan.PlanWritables;
 import org.elasticsearch.xpack.esql.querydsl.query.SingleValueQuery;
 import org.elasticsearch.xpack.esql.querylog.EsqlQueryLog;
@@ -262,7 +264,8 @@ public class EsqlPlugin extends Plugin implements ActionPlugin, ExtensiblePlugin
             ESQL_QUERYLOG_THRESHOLD_WARN_SETTING,
             ESQL_QUERYLOG_INCLUDE_USER_SETTING,
             DEFAULT_DATA_PARTITIONING,
-            STORED_FIELDS_SEQUENTIAL_PROPORTION
+            STORED_FIELDS_SEQUENTIAL_PROPORTION,
+            EsqlFlags.ESQL_STRING_LIKE_ON_INDEX
         );
     }
 
@@ -325,6 +328,8 @@ public class EsqlPlugin extends Plugin implements ActionPlugin, ExtensiblePlugin
         entries.add(AsyncOperator.Status.ENTRY);
         entries.add(EnrichLookupOperator.Status.ENTRY);
         entries.add(LookupFromIndexOperator.Status.ENTRY);
+        entries.add(ExpressionQueryBuilder.ENTRY);
+        entries.add(PlanStreamWrapperQueryBuilder.ENTRY);
 
         entries.addAll(ExpressionWritables.getNamedWriteables());
         entries.addAll(PlanWritables.getNamedWriteables());

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

@@ -174,6 +174,7 @@ public class TransportEsqlQueryAction extends HandledTransportAction<EsqlQueryRe
         defaultAllowPartialResults = EsqlPlugin.QUERY_ALLOW_PARTIAL_RESULTS.get(clusterService.getSettings());
         clusterService.getClusterSettings()
             .addSettingsUpdateConsumer(EsqlPlugin.QUERY_ALLOW_PARTIAL_RESULTS, v -> defaultAllowPartialResults = v);
+
     }
 
     @Override
@@ -213,6 +214,7 @@ public class TransportEsqlQueryAction extends HandledTransportAction<EsqlQueryRe
         if (request.allowPartialResults() == null) {
             request.allowPartialResults(defaultAllowPartialResults);
         }
+        EsqlFlags flags = computeService.createFlags();
         Configuration configuration = new Configuration(
             ZoneOffset.UTC,
             request.locale() != null ? request.locale() : Locale.US,
@@ -236,6 +238,7 @@ public class TransportEsqlQueryAction extends HandledTransportAction<EsqlQueryRe
         PlanRunner planRunner = (plan, resultListener) -> computeService.execute(
             sessionId,
             (CancellableTask) task,
+            flags,
             plan,
             configuration,
             foldCtx,

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/querydsl/query/EqualsSyntheticSourceDelegate.java

@@ -35,6 +35,11 @@ public class EqualsSyntheticSourceDelegate extends Query {
         return fieldName + "(delegate):" + value;
     }
 
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
+
     private class Builder extends BaseTermQueryBuilder<Builder> {
         private Builder(String name, String value) {
             super(name, value);

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/querydsl/query/KnnQuery.java

@@ -81,4 +81,9 @@ public class KnnQuery extends Query {
     public boolean scorable() {
         return true;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/querydsl/query/KqlQuery.java

@@ -87,4 +87,9 @@ public class KqlQuery extends Query {
     public boolean scorable() {
         return true;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/querydsl/query/MatchPhraseQuery.java

@@ -108,4 +108,9 @@ public class MatchPhraseQuery extends Query {
     public boolean scorable() {
         return true;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/querydsl/query/MatchQuery.java

@@ -130,4 +130,9 @@ public class MatchQuery extends Query {
     public boolean scorable() {
         return true;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/querydsl/query/MultiMatchQuery.java

@@ -114,4 +114,9 @@ public class MultiMatchQuery extends Query {
     public boolean scorable() {
         return true;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/querydsl/query/SingleValueQuery.java

@@ -125,6 +125,11 @@ public class SingleValueQuery extends Query {
         return Objects.hash(super.hashCode(), next, field, useSyntheticSourceDelegate);
     }
 
+    @Override
+    public boolean containsPlan() {
+        return next.containsPlan();
+    }
+
     public abstract static class AbstractBuilder extends AbstractQueryBuilder<AbstractBuilder> {
         private final QueryBuilder next;
         private final String field;

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/querydsl/query/SpatialRelatesQuery.java

@@ -86,6 +86,11 @@ public class SpatialRelatesQuery extends Query {
         };
     }
 
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
+
     /**
      * This class is a minimal implementation of the QueryBuilder interface.
      * We only need the toQuery method, but ESQL makes extensive use of QueryBuilder and trimming that interface down for ESQL only would

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/querydsl/query/TranslationAwareExpressionQuery.java

@@ -40,4 +40,9 @@ public class TranslationAwareExpressionQuery extends Query {
         // All Full Text Functions are translated to queries using this method
         return true;
     }
+
+    @Override
+    public boolean containsPlan() {
+        return false;
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/Configuration.java

@@ -211,6 +211,23 @@ public class Configuration implements Writeable {
         return tables;
     }
 
+    public Configuration withoutTables() {
+        return new Configuration(
+            zoneId,
+            locale,
+            username,
+            clusterName,
+            pragmas,
+            resultTruncationMaxSize,
+            resultTruncationDefaultSize,
+            query,
+            profile,
+            Map.of(),
+            queryStartTimeNanos,
+            allowPartialResults
+        );
+    }
+
     /**
      * Enable profiling, sacrificing performance to return information about
      * what operations are taking the most time.
@@ -309,4 +326,11 @@ public class Configuration implements Writeable {
             + '}';
     }
 
+    /**
+     * Reads a {@link Configuration} that doesn't contain any {@link Configuration#tables()}.
+     */
+    public static Configuration readWithoutTables(StreamInput in) throws IOException {
+        BlockStreamInput blockStreamInput = new BlockStreamInput(in, null);
+        return new Configuration(blockStreamInput);
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java

@@ -326,7 +326,7 @@ public class EsqlSession {
     }
 
     private LogicalPlan parse(String query, QueryParams params) {
-        var parsed = new EsqlParser().createStatement(query, params, planTelemetry);
+        var parsed = new EsqlParser().createStatement(query, params, planTelemetry, configuration);
         LOGGER.debug("Parsed logical plan:\n{}", parsed);
         return parsed;
     }

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

@@ -88,6 +88,7 @@ import org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner.LocalExecution
 import org.elasticsearch.xpack.esql.planner.PlannerUtils;
 import org.elasticsearch.xpack.esql.planner.TestPhysicalOperationProviders;
 import org.elasticsearch.xpack.esql.planner.mapper.Mapper;
+import org.elasticsearch.xpack.esql.plugin.EsqlFlags;
 import org.elasticsearch.xpack.esql.plugin.QueryPragmas;
 import org.elasticsearch.xpack.esql.session.Configuration;
 import org.elasticsearch.xpack.esql.session.EsqlSession;
@@ -566,7 +567,7 @@ public class CsvTests extends ESTestCase {
     }
 
     private ActualResults executePlan(BigArrays bigArrays) throws Exception {
-        LogicalPlan parsed = parser.createStatement(testCase.query);
+        LogicalPlan parsed = parser.createStatement(testCase.query, EsqlTestUtils.TEST_CFG);
         var testDatasets = testDatasets(parsed);
         LogicalPlan analyzed = analyzedPlan(parsed, testDatasets);
 
@@ -715,7 +716,7 @@ public class CsvTests extends ESTestCase {
             var searchStats = new DisabledSearchStats();
             var logicalTestOptimizer = new LocalLogicalPlanOptimizer(new LocalLogicalOptimizerContext(configuration, foldCtx, searchStats));
             var physicalTestOptimizer = new TestLocalPhysicalPlanOptimizer(
-                new LocalPhysicalOptimizerContext(configuration, foldCtx, searchStats)
+                new LocalPhysicalOptimizerContext(new EsqlFlags(true), configuration, foldCtx, searchStats)
             );
 
             var csvDataNodePhysicalPlan = PlannerUtils.localPlan(dataNodePlan, logicalTestOptimizer, physicalTestOptimizer);

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

@@ -129,7 +129,7 @@ public final class AnalyzerTestUtils {
     }
 
     public static LogicalPlan analyze(String query, Analyzer analyzer) {
-        var plan = new EsqlParser().createStatement(query);
+        var plan = new EsqlParser().createStatement(query, configuration(query));
         // System.out.println(plan);
         var analyzed = analyzer.analyze(plan);
         // System.out.println(analyzed);
@@ -137,7 +137,7 @@ public final class AnalyzerTestUtils {
     }
 
     public static LogicalPlan analyze(String query, String mapping, QueryParams params) {
-        var plan = new EsqlParser().createStatement(query, params);
+        var plan = new EsqlParser().createStatement(query, params, configuration(query));
         var analyzer = analyzer(loadMapping(mapping, "test"), TEST_VERIFIER, configuration(query));
         return analyzer.analyze(plan);
     }

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

@@ -99,7 +99,7 @@ public class ParsingTests extends ESTestCase {
                 if (EsqlDataTypeConverter.converterFunctionFactory(expectedType) == null) {
                     continue;
                 }
-                LogicalPlan plan = parser.createStatement("ROW a = 1::" + nameOrAlias);
+                LogicalPlan plan = parser.createStatement("ROW a = 1::" + nameOrAlias, TEST_CFG);
                 Row row = as(plan, Row.class);
                 assertThat(row.fields(), hasSize(1));
                 Function functionCall = (Function) row.fields().get(0).child();
@@ -210,7 +210,10 @@ public class ParsingTests extends ESTestCase {
     }
 
     private String error(String query, QueryParams params) {
-        ParsingException e = expectThrows(ParsingException.class, () -> defaultAnalyzer.analyze(parser.createStatement(query, params)));
+        ParsingException e = expectThrows(
+            ParsingException.class,
+            () -> defaultAnalyzer.analyze(parser.createStatement(query, params, TEST_CFG))
+        );
         String message = e.getMessage();
         assertTrue(message.startsWith("line "));
         return message.substring("line ".length());

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

@@ -37,6 +37,7 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 
+import static org.elasticsearch.xpack.esql.EsqlTestUtils.TEST_CFG;
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.paramAsConstant;
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.withDefaultLimitWarning;
 import static org.elasticsearch.xpack.esql.analysis.AnalyzerTestUtils.loadMapping;
@@ -2239,7 +2240,7 @@ public class VerifierTests extends ESTestCase {
     }
 
     private void query(String query, Analyzer analyzer) {
-        analyzer.analyze(parser.createStatement(query));
+        analyzer.analyze(parser.createStatement(query, TEST_CFG));
     }
 
     private String error(String query) {
@@ -2270,7 +2271,7 @@ public class VerifierTests extends ESTestCase {
         Throwable e = expectThrows(
             exception,
             "Expected error for query [" + query + "] but no error was raised",
-            () -> analyzer.analyze(parser.createStatement(query, new QueryParams(parameters)))
+            () -> analyzer.analyze(parser.createStatement(query, new QueryParams(parameters), TEST_CFG))
         );
         assertThat(e, instanceOf(exception));
 

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

@@ -79,7 +79,7 @@ public class CheckLicenseTests extends ESTestCase {
             }
         };
 
-        var plan = parser.createStatement(esql);
+        var plan = parser.createStatement(esql, EsqlTestUtils.TEST_CFG);
         plan = plan.transformDown(
             Limit.class,
             l -> Objects.equals(l.limit().fold(FoldContext.small()), 10)

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

@@ -134,6 +134,6 @@ public class EndsWithTests extends AbstractScalarFunctionTestCase {
 
         var query = function.asQuery(LucenePushdownPredicates.DEFAULT, TranslatorHandler.TRANSLATOR_HANDLER);
 
-        assertThat(query, equalTo(new WildcardQuery(Source.EMPTY, "field", "*a\\*b\\?c\\\\")));
+        assertThat(query, equalTo(new WildcardQuery(Source.EMPTY, "field", "*a\\*b\\?c\\\\", false, false)));
     }
 }

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

@@ -94,6 +94,6 @@ public class StartsWithTests extends AbstractScalarFunctionTestCase {
 
         var query = function.asQuery(LucenePushdownPredicates.DEFAULT, TranslatorHandler.TRANSLATOR_HANDLER);
 
-        assertThat(query, equalTo(new WildcardQuery(Source.EMPTY, "field", "a\\*b\\?c\\\\*")));
+        assertThat(query, equalTo(new WildcardQuery(Source.EMPTY, "field", "a\\*b\\?c\\\\*", false, false)));
     }
 }

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

@@ -11,7 +11,6 @@ import com.carrotsearch.randomizedtesting.annotations.Name;
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 
 import org.apache.lucene.util.BytesRef;
-import org.elasticsearch.TransportVersion;
 import org.elasticsearch.xpack.esql.EsqlTestUtils;
 import org.elasticsearch.xpack.esql.capabilities.TranslationAware;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
@@ -26,12 +25,14 @@ import org.elasticsearch.xpack.esql.expression.function.FunctionName;
 import org.elasticsearch.xpack.esql.expression.function.TestCaseSupplier;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.regex.WildcardLikeList;
 import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.LucenePushdownPredicates;
+import org.elasticsearch.xpack.esql.plugin.EsqlFlags;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
+import static org.elasticsearch.TransportVersions.V_8_17_0;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.startsWith;
 
@@ -104,7 +105,7 @@ public class WildcardLikeListTests extends AbstractScalarFunctionTestCase {
     public void testNotPushableOverCanMatch() {
         TranslationAware translatable = (TranslationAware) buildFieldExpression(testCase);
         assertThat(
-            translatable.translatable(LucenePushdownPredicates.forCanMatch(TransportVersion.current())).finish(),
+            translatable.translatable(LucenePushdownPredicates.forCanMatch(V_8_17_0, new EsqlFlags(true))).finish(),
             equalTo(TranslationAware.FinishedTranslatable.NO)
         );
     }
@@ -112,7 +113,7 @@ public class WildcardLikeListTests extends AbstractScalarFunctionTestCase {
     public void testPushable() {
         TranslationAware translatable = (TranslationAware) buildFieldExpression(testCase);
         assertThat(
-            translatable.translatable(LucenePushdownPredicates.from(new EsqlTestUtils.TestSearchStats())).finish(),
+            translatable.translatable(LucenePushdownPredicates.from(new EsqlTestUtils.TestSearchStats(), new EsqlFlags(true))).finish(),
             equalTo(TranslationAware.FinishedTranslatable.YES)
         );
     }

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

@@ -83,7 +83,7 @@ public class WildcardLikeTests extends AbstractScalarFunctionTestCase {
         return buildWildcardLike(source, args);
     }
 
-    static Expression buildWildcardLike(Source source, List<Expression> args) {
+    Expression buildWildcardLike(Source source, List<Expression> args) {
         Expression expression = args.get(0);
         Literal pattern = (Literal) args.get(1);
         Literal caseInsensitive = args.size() > 2 ? (Literal) args.get(2) : null;

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/AbstractLogicalPlanOptimizerTests.java

@@ -180,7 +180,7 @@ public abstract class AbstractLogicalPlanOptimizerTests extends ESTestCase {
     }
 
     protected LogicalPlan plan(String query, LogicalPlanOptimizer optimizer) {
-        var analyzed = analyzer.analyze(parser.createStatement(query));
+        var analyzed = analyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG));
         // System.out.println(analyzed);
         var optimized = optimizer.optimize(analyzed);
         // System.out.println(optimized);
@@ -188,7 +188,7 @@ public abstract class AbstractLogicalPlanOptimizerTests extends ESTestCase {
     }
 
     protected LogicalPlan planAirports(String query) {
-        var analyzed = analyzerAirports.analyze(parser.createStatement(query));
+        var analyzed = analyzerAirports.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG));
         // System.out.println(analyzed);
         var optimized = logicalOptimizer.optimize(analyzed);
         // System.out.println(optimized);
@@ -196,7 +196,7 @@ public abstract class AbstractLogicalPlanOptimizerTests extends ESTestCase {
     }
 
     protected LogicalPlan planExtra(String query) {
-        var analyzed = analyzerExtra.analyze(parser.createStatement(query));
+        var analyzed = analyzerExtra.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG));
         // System.out.println(analyzed);
         var optimized = logicalOptimizer.optimize(analyzed);
         // System.out.println(optimized);
@@ -204,11 +204,11 @@ public abstract class AbstractLogicalPlanOptimizerTests extends ESTestCase {
     }
 
     protected LogicalPlan planTypes(String query) {
-        return logicalOptimizer.optimize(analyzerTypes.analyze(parser.createStatement(query)));
+        return logicalOptimizer.optimize(analyzerTypes.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
     }
 
     protected LogicalPlan planMultiIndex(String query) {
-        return logicalOptimizer.optimize(multiIndexAnalyzer.analyze(parser.createStatement(query)));
+        return logicalOptimizer.optimize(multiIndexAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
     }
 
     @Override

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LocalLogicalPlanOptimizerTests.java

@@ -508,7 +508,7 @@ public class LocalLogicalPlanOptimizerTests extends ESTestCase {
             TEST_VERIFIER
         );
 
-        var analyzed = analyzer.analyze(parser.createStatement(query));
+        var analyzed = analyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG));
         var optimized = logicalOptimizer.optimize(analyzed);
         var localContext = new LocalLogicalOptimizerContext(EsqlTestUtils.TEST_CFG, FoldContext.small(), searchStats);
         var plan = new LocalLogicalPlanOptimizer(localContext).localOptimize(optimized);
@@ -785,7 +785,7 @@ public class LocalLogicalPlanOptimizerTests extends ESTestCase {
     }
 
     private LogicalPlan plan(String query, Analyzer analyzer) {
-        var analyzed = analyzer.analyze(parser.createStatement(query));
+        var analyzed = analyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG));
         // System.out.println(analyzed);
         var optimized = logicalOptimizer.optimize(analyzed);
         // System.out.println(optimized);

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java

@@ -5254,20 +5254,26 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
             TEST_VERIFIER
         );
 
-        var plan = logicalOptimizer.optimize(analyzer.analyze(parser.createStatement("from empty_test")));
+        var plan = logicalOptimizer.optimize(analyzer.analyze(parser.createStatement("from empty_test", EsqlTestUtils.TEST_CFG)));
         as(plan, LocalRelation.class);
         assertThat(plan.output(), equalTo(NO_FIELDS));
 
-        plan = logicalOptimizer.optimize(analyzer.analyze(parser.createStatement("from empty_test metadata _id | eval x = 1")));
+        plan = logicalOptimizer.optimize(
+            analyzer.analyze(parser.createStatement("from empty_test metadata _id | eval x = 1", EsqlTestUtils.TEST_CFG))
+        );
         as(plan, LocalRelation.class);
         assertThat(Expressions.names(plan.output()), contains("_id", "x"));
 
-        plan = logicalOptimizer.optimize(analyzer.analyze(parser.createStatement("from empty_test metadata _id, _version | limit 5")));
+        plan = logicalOptimizer.optimize(
+            analyzer.analyze(parser.createStatement("from empty_test metadata _id, _version | limit 5", EsqlTestUtils.TEST_CFG))
+        );
         as(plan, LocalRelation.class);
         assertThat(Expressions.names(plan.output()), contains("_id", "_version"));
 
         plan = logicalOptimizer.optimize(
-            analyzer.analyze(parser.createStatement("from empty_test | eval x = \"abc\" | enrich languages_idx on x"))
+            analyzer.analyze(
+                parser.createStatement("from empty_test | eval x = \"abc\" | enrich languages_idx on x", EsqlTestUtils.TEST_CFG)
+            )
         );
         LocalRelation local = as(plan, LocalRelation.class);
         assertThat(Expressions.names(local.output()), contains(NO_FIELDS.get(0).name(), "x", "language_code", "language_name"));
@@ -5962,7 +5968,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
 
     private void assertSemanticMatching(String expected, String provided) {
         BinaryComparison bc = extractPlannedBinaryComparison(provided);
-        LogicalPlan exp = analyzerTypes.analyze(parser.createStatement("FROM types | WHERE " + expected));
+        LogicalPlan exp = analyzerTypes.analyze(parser.createStatement("FROM types | WHERE " + expected, EsqlTestUtils.TEST_CFG));
         assertSemanticMatching(bc, extractPlannedBinaryComparison(exp));
     }
 
@@ -5990,7 +5996,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
     private void assertNotSimplified(String comparison) {
         String query = "FROM types | WHERE " + comparison;
         Expression optimized = getComparisonFromLogicalPlan(planTypes(query));
-        Expression raw = getComparisonFromLogicalPlan(analyzerTypes.analyze(parser.createStatement(query)));
+        Expression raw = getComparisonFromLogicalPlan(analyzerTypes.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
 
         assertTrue(raw.semanticEquals(optimized));
     }
@@ -6683,7 +6689,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
     public void testTranslateMetricsWithoutGrouping() {
         assumeTrue("requires snapshot builds", Build.current().isSnapshot());
         var query = "TS k8s | STATS max(rate(network.total_bytes_in))";
-        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         Limit limit = as(plan, Limit.class);
         Aggregate finalAggs = as(limit.child(), Aggregate.class);
         assertThat(finalAggs, not(instanceOf(TimeSeriesAggregate.class)));
@@ -6704,7 +6710,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
     public void testTranslateMixedAggsWithoutGrouping() {
         assumeTrue("requires snapshot builds", Build.current().isSnapshot());
         var query = "TS k8s | STATS max(rate(network.total_bytes_in)), max(network.cost)";
-        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         Limit limit = as(plan, Limit.class);
         Aggregate finalAggs = as(limit.child(), Aggregate.class);
         assertThat(finalAggs, not(instanceOf(TimeSeriesAggregate.class)));
@@ -6729,7 +6735,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
     public void testTranslateMixedAggsWithMathWithoutGrouping() {
         assumeTrue("requires snapshot builds", Build.current().isSnapshot());
         var query = "TS k8s | STATS max(rate(network.total_bytes_in)), max(network.cost + 0.2) * 1.1";
-        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         Project project = as(plan, Project.class);
         Eval mulEval = as(project.child(), Eval.class);
         assertThat(mulEval.fields(), hasSize(1));
@@ -6767,7 +6773,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
     public void testTranslateMetricsGroupedByOneDimension() {
         assumeTrue("requires snapshot builds", Build.current().isSnapshot());
         var query = "TS k8s | STATS sum(rate(network.total_bytes_in)) BY cluster | SORT cluster | LIMIT 10";
-        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         TopN topN = as(plan, TopN.class);
         Aggregate aggsByCluster = as(topN.child(), Aggregate.class);
         assertThat(aggsByCluster, not(instanceOf(TimeSeriesAggregate.class)));
@@ -6792,7 +6798,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
     public void testTranslateMetricsGroupedByTwoDimension() {
         assumeTrue("requires snapshot builds", Build.current().isSnapshot());
         var query = "TS k8s | STATS avg(rate(network.total_bytes_in)) BY cluster, pod";
-        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         Project project = as(plan, Project.class);
         Eval eval = as(project.child(), Eval.class);
         assertThat(eval.fields(), hasSize(1));
@@ -6832,7 +6838,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
     public void testTranslateMetricsGroupedByTimeBucket() {
         assumeTrue("requires snapshot builds", Build.current().isSnapshot());
         var query = "TS k8s | STATS sum(rate(network.total_bytes_in)) BY bucket(@timestamp, 1h)";
-        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         Limit limit = as(plan, Limit.class);
         Aggregate finalAgg = as(limit.child(), Aggregate.class);
         assertThat(finalAgg, not(instanceOf(TimeSeriesAggregate.class)));
@@ -6866,7 +6872,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
             | SORT cluster
             | LIMIT 10
             """;
-        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         Project project = as(plan, Project.class);
         TopN topN = as(project.child(), TopN.class);
         Eval eval = as(topN.child(), Eval.class);
@@ -6908,7 +6914,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
             | SORT cluster
             | LIMIT 10
             """;
-        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         TopN topN = as(plan, TopN.class);
         Aggregate finalAgg = as(topN.child(), Aggregate.class);
         Eval eval = as(finalAgg.child(), Eval.class);
@@ -6929,7 +6935,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
             | SORT cluster
             | LIMIT 10
             """;
-        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         Project project = as(plan, Project.class);
         TopN topN = as(project.child(), TopN.class);
         Eval eval = as(topN.child(), Eval.class);
@@ -6981,7 +6987,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
             | SORT cluster
             | LIMIT 10
             """;
-        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         Project project = as(plan, Project.class);
         TopN topN = as(project.child(), TopN.class);
         Eval evalDiv = as(topN.child(), Eval.class);
@@ -7034,7 +7040,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
     public void testTranslateMaxOverTime() {
         assumeTrue("requires snapshot builds", Build.current().isSnapshot());
         var query = "TS k8s | STATS sum(max_over_time(network.bytes_in)) BY bucket(@timestamp, 1h)";
-        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         Limit limit = as(plan, Limit.class);
         Aggregate finalAgg = as(limit.child(), Aggregate.class);
         assertThat(finalAgg, not(instanceOf(TimeSeriesAggregate.class)));
@@ -7063,7 +7069,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
     public void testTranslateAvgOverTime() {
         assumeTrue("requires snapshot builds", Build.current().isSnapshot());
         var query = "TS k8s | STATS sum(avg_over_time(network.bytes_in)) BY bucket(@timestamp, 1h)";
-        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+        var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         Limit limit = as(plan, Limit.class);
         Aggregate finalAgg = as(limit.child(), Aggregate.class);
         assertThat(finalAgg, not(instanceOf(TimeSeriesAggregate.class)));
@@ -7103,7 +7109,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
             """);
         List<LogicalPlan> plans = new ArrayList<>();
         for (String query : queries) {
-            var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query)));
+            var plan = logicalOptimizer.optimize(metricsAnalyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
             plans.add(plan);
         }
         for (LogicalPlan plan : plans) {
@@ -7694,7 +7700,7 @@ public class LogicalPlanOptimizerTests extends AbstractLogicalPlanOptimizerTests
             | mv_expand x
             | sort y
             """;
-        LogicalPlan analyzed = analyzer.analyze(parser.createStatement(query));
+        LogicalPlan analyzed = analyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG));
         LogicalPlan optimized = rule.apply(analyzed);
 
         // check that all the redundant SORTs are removed in a single run

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/OptimizerRulesTests.java

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.esql.optimizer;
 
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.esql.EsqlTestUtils;
 import org.elasticsearch.xpack.esql.core.expression.Alias;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
@@ -128,7 +129,7 @@ public class OptimizerRulesTests extends ESTestCase {
         };
 
         rule.apply(
-            new EsqlParser().createStatement("FROM index | EVAL x=f1+1 | KEEP x, f2 | LIMIT 1"),
+            new EsqlParser().createStatement("FROM index | EVAL x=f1+1 | KEEP x, f2 | LIMIT 1", EsqlTestUtils.TEST_CFG),
             new LogicalOptimizerContext(null, FoldContext.small())
         );
 

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/PhysicalPlanOptimizerTests.java

@@ -136,6 +136,7 @@ import org.elasticsearch.xpack.esql.planner.EsPhysicalOperationProviders;
 import org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner;
 import org.elasticsearch.xpack.esql.planner.PlannerUtils;
 import org.elasticsearch.xpack.esql.planner.mapper.Mapper;
+import org.elasticsearch.xpack.esql.plugin.EsqlFlags;
 import org.elasticsearch.xpack.esql.plugin.QueryPragmas;
 import org.elasticsearch.xpack.esql.querydsl.query.EqualsSyntheticSourceDelegate;
 import org.elasticsearch.xpack.esql.querydsl.query.SingleValueQuery;
@@ -7842,7 +7843,7 @@ public class PhysicalPlanOptimizerTests extends ESTestCase {
         // The TopN needs an estimated row size for the planner to work
         var plans = PlannerUtils.breakPlanBetweenCoordinatorAndDataNode(EstimatesRowSize.estimateRowSize(0, plan), config);
         plan = useDataNodePlan ? plans.v2() : plans.v1();
-        plan = PlannerUtils.localPlan(config, FoldContext.small(), plan, TEST_SEARCH_STATS);
+        plan = PlannerUtils.localPlan(new EsqlFlags(true), config, FoldContext.small(), plan, TEST_SEARCH_STATS);
         ExchangeSinkHandler exchangeSinkHandler = new ExchangeSinkHandler(null, 10, () -> 10);
         LocalExecutionPlanner planner = new LocalExecutionPlanner(
             "test",
@@ -8209,7 +8210,7 @@ public class PhysicalPlanOptimizerTests extends ESTestCase {
         // individually hence why here the plan is kept as is
 
         var l = p.transformUp(FragmentExec.class, fragment -> {
-            var localPlan = PlannerUtils.localPlan(config, FoldContext.small(), fragment, searchStats);
+            var localPlan = PlannerUtils.localPlan(new EsqlFlags(true), config, FoldContext.small(), fragment, searchStats);
             return EstimatesRowSize.estimateRowSize(fragment.estimatedRowSize(), localPlan);
         });
 
@@ -8247,7 +8248,7 @@ public class PhysicalPlanOptimizerTests extends ESTestCase {
     }
 
     private PhysicalPlan physicalPlan(String query, TestDataSource dataSource, boolean assertSerialization) {
-        var logical = logicalOptimizer.optimize(dataSource.analyzer.analyze(parser.createStatement(query)));
+        var logical = logicalOptimizer.optimize(dataSource.analyzer.analyze(parser.createStatement(query, config)));
         // System.out.println("Logical\n" + logical);
         var physical = mapper.map(logical);
         // System.out.println("Physical\n" + physical);

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/TestPlannerOptimizer.java

@@ -15,6 +15,7 @@ import org.elasticsearch.xpack.esql.plan.physical.EstimatesRowSize;
 import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
 import org.elasticsearch.xpack.esql.planner.PlannerUtils;
 import org.elasticsearch.xpack.esql.planner.mapper.Mapper;
+import org.elasticsearch.xpack.esql.plugin.EsqlFlags;
 import org.elasticsearch.xpack.esql.session.Configuration;
 import org.elasticsearch.xpack.esql.stats.SearchStats;
 
@@ -66,7 +67,7 @@ public class TestPlannerOptimizer {
             new LocalLogicalOptimizerContext(config, FoldContext.small(), searchStats)
         );
         var physicalTestOptimizer = new TestLocalPhysicalPlanOptimizer(
-            new LocalPhysicalOptimizerContext(config, FoldContext.small(), searchStats),
+            new LocalPhysicalOptimizerContext(new EsqlFlags(true), config, FoldContext.small(), searchStats),
             true
         );
         var l = PlannerUtils.localPlan(physicalPlan, logicalTestOptimizer, physicalTestOptimizer);
@@ -79,7 +80,7 @@ public class TestPlannerOptimizer {
     }
 
     private PhysicalPlan physicalPlan(String query, Analyzer analyzer) {
-        var logical = logicalOptimizer.optimize(analyzer.analyze(parser.createStatement(query)));
+        var logical = logicalOptimizer.optimize(analyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         // System.out.println("Logical\n" + logical);
         var physical = mapper.map(logical);
         return physical;

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PropagateInlineEvalsTests.java

@@ -169,7 +169,7 @@ public class PropagateInlineEvalsTests extends ESTestCase {
     }
 
     private LogicalPlan plan(String query, LogicalPlanOptimizer optimizer) {
-        return optimizer.optimize(analyzer.analyze(parser.createStatement(query)));
+        return optimizer.optimize(analyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
     }
 
     @Override

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/rules/physical/local/PushTopNToSourceTests.java

@@ -35,6 +35,7 @@ import org.elasticsearch.xpack.esql.plan.physical.EsQueryExec;
 import org.elasticsearch.xpack.esql.plan.physical.EvalExec;
 import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
 import org.elasticsearch.xpack.esql.plan.physical.TopNExec;
+import org.elasticsearch.xpack.esql.plugin.EsqlFlags;
 import org.elasticsearch.xpack.esql.stats.SearchStats;
 
 import java.io.IOException;
@@ -416,7 +417,7 @@ public class PushTopNToSourceTests extends ESTestCase {
 
     private static PhysicalPlan pushTopNToSource(TopNExec topNExec) {
         var configuration = EsqlTestUtils.configuration("from test");
-        var ctx = new LocalPhysicalOptimizerContext(configuration, FoldContext.small(), SearchStats.EMPTY);
+        var ctx = new LocalPhysicalOptimizerContext(new EsqlFlags(true), configuration, FoldContext.small(), SearchStats.EMPTY);
         var pushTopNToSource = new PushTopNToSource();
         return pushTopNToSource.rule(topNExec, ctx);
     }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/AbstractStatementParserTests.java

@@ -11,6 +11,7 @@ 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.EsqlTestUtils;
 import org.elasticsearch.xpack.esql.VerificationException;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
@@ -57,11 +58,11 @@ abstract class AbstractStatementParserTests extends ESTestCase {
     }
 
     LogicalPlan statement(String e, QueryParams params) {
-        return parser.createStatement(e, params);
+        return parser.createStatement(e, params, EsqlTestUtils.TEST_CFG);
     }
 
     LogicalPlan processingCommand(String e) {
-        return parser.createStatement("row a = 1 | " + e);
+        return parser.createStatement("row a = 1 | " + e, EsqlTestUtils.TEST_CFG);
     }
 
     static UnresolvedAttribute attribute(String name) {
@@ -170,7 +171,7 @@ abstract class AbstractStatementParserTests extends ESTestCase {
             "Query [" + query + "] is expected to throw " + VerificationException.class + " with message [" + errorMessage + "]",
             VerificationException.class,
             containsString(errorMessage),
-            () -> parser.createStatement(query)
+            () -> parser.createStatement(query, EsqlTestUtils.TEST_CFG)
         );
     }
 

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/ExpressionTests.java

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.esql.parser;
 
 import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.esql.EsqlTestUtils;
 import org.elasticsearch.xpack.esql.core.expression.Alias;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.FoldContext;
@@ -659,7 +660,7 @@ public class ExpressionTests extends ESTestCase {
     }
 
     private LogicalPlan parse(String s) {
-        return parser.createStatement(s);
+        return parser.createStatement(s, EsqlTestUtils.TEST_CFG);
     }
 
     private Literal l(Object value, DataType type) {

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/GrammarInDevelopmentParsingTests.java

@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.esql.parser;
 
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.esql.EsqlTestUtils;
 
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.not;
@@ -35,7 +36,7 @@ public class GrammarInDevelopmentParsingTests extends ESTestCase {
     }
 
     void parse(String query, String errorMessage) {
-        ParsingException pe = expectThrows(ParsingException.class, () -> parser().createStatement(query));
+        ParsingException pe = expectThrows(ParsingException.class, () -> parser().createStatement(query, EsqlTestUtils.TEST_CFG));
         assertThat(pe.getMessage(), containsString("mismatched input '" + errorMessage + "'"));
         // check the parser eliminated the DEV_ tokens from the message
         assertThat(pe.getMessage(), not(containsString("DEV_")));

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/StatementParserTests.java

@@ -13,6 +13,7 @@ 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.EsqlTestUtils;
 import org.elasticsearch.xpack.esql.action.EsqlCapabilities;
 import org.elasticsearch.xpack.esql.core.capabilities.UnresolvedException;
 import org.elasticsearch.xpack.esql.core.expression.Alias;
@@ -3835,7 +3836,7 @@ public class StatementParserTests extends AbstractStatementParserTests {
             List.of(paramAsConstant(null, "query text"), paramAsConstant(null, "reranker"), paramAsConstant(null, "rerank_score"))
         );
         var rerank = as(
-            parser.createStatement("row a = 1 | RERANK ? ON title WITH inferenceId=?, scoreColumn=? ", queryParams),
+            parser.createStatement("row a = 1 | RERANK ? ON title WITH inferenceId=?, scoreColumn=? ", queryParams, EsqlTestUtils.TEST_CFG),
             Rerank.class
         );
 
@@ -3858,7 +3859,8 @@ public class StatementParserTests extends AbstractStatementParserTests {
         var rerank = as(
             parser.createStatement(
                 "row a = 1 | RERANK ?queryText ON title WITH inferenceId=?inferenceId, scoreColumn=?scoreColumnName",
-                queryParams
+                queryParams,
+                EsqlTestUtils.TEST_CFG
             ),
             Rerank.class
         );
@@ -3906,7 +3908,10 @@ public class StatementParserTests extends AbstractStatementParserTests {
 
     public void testCompletionWithPositionalParameters() {
         var queryParams = new QueryParams(List.of(paramAsConstant(null, "inferenceId")));
-        var plan = as(parser.createStatement("row a = 1 | COMPLETION prompt_field WITH ?", queryParams), Completion.class);
+        var plan = as(
+            parser.createStatement("row a = 1 | COMPLETION prompt_field WITH ?", queryParams, EsqlTestUtils.TEST_CFG),
+            Completion.class
+        );
 
         assertThat(plan.prompt(), equalTo(attribute("prompt_field")));
         assertThat(plan.inferenceId(), equalTo(literalString("inferenceId")));
@@ -3915,7 +3920,10 @@ public class StatementParserTests extends AbstractStatementParserTests {
 
     public void testCompletionWithNamedParameters() {
         var queryParams = new QueryParams(List.of(paramAsConstant("inferenceId", "myInference")));
-        var plan = as(parser.createStatement("row a = 1 | COMPLETION prompt_field WITH ?inferenceId", queryParams), Completion.class);
+        var plan = as(
+            parser.createStatement("row a = 1 | COMPLETION prompt_field WITH ?inferenceId", queryParams, EsqlTestUtils.TEST_CFG),
+            Completion.class
+        );
 
         assertThat(plan.prompt(), equalTo(attribute("prompt_field")));
         assertThat(plan.inferenceId(), equalTo(literalString("myInference")));

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/planner/FilterTests.java

@@ -16,7 +16,6 @@ import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.logging.LoggerMessageFormat;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.query.AbstractQueryBuilder;
-import org.elasticsearch.index.query.AutomatonQueryBuilder;
 import org.elasticsearch.index.query.BoolQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.test.ESTestCase;
@@ -30,8 +29,10 @@ import org.elasticsearch.xpack.esql.core.util.Queries;
 import org.elasticsearch.xpack.esql.expression.function.EsqlFunctionRegistry;
 import org.elasticsearch.xpack.esql.index.EsIndex;
 import org.elasticsearch.xpack.esql.index.IndexResolution;
+import org.elasticsearch.xpack.esql.io.stream.ExpressionQueryBuilder;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput;
+import org.elasticsearch.xpack.esql.io.stream.PlanStreamWrapperQueryBuilder;
 import org.elasticsearch.xpack.esql.optimizer.LogicalPlanOptimizer;
 import org.elasticsearch.xpack.esql.optimizer.PhysicalOptimizerContext;
 import org.elasticsearch.xpack.esql.optimizer.PhysicalPlanOptimizer;
@@ -39,6 +40,7 @@ import org.elasticsearch.xpack.esql.parser.EsqlParser;
 import org.elasticsearch.xpack.esql.plan.physical.FragmentExec;
 import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
 import org.elasticsearch.xpack.esql.planner.mapper.Mapper;
+import org.elasticsearch.xpack.esql.plugin.EsqlFlags;
 import org.elasticsearch.xpack.esql.querydsl.query.SingleValueQuery;
 import org.elasticsearch.xpack.esql.session.Configuration;
 import org.junit.BeforeClass;
@@ -50,6 +52,7 @@ import java.util.Map;
 import java.util.Set;
 
 import static java.util.Arrays.asList;
+import static org.elasticsearch.TransportVersions.V_8_17_0;
 import static org.elasticsearch.index.query.QueryBuilders.rangeQuery;
 import static org.elasticsearch.xpack.esql.ConfigurationTestUtils.randomConfiguration;
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.TEST_VERIFIER;
@@ -324,8 +327,8 @@ public class FilterTests extends ESTestCase {
             |WHERE {} LIKE ("a+", "b+")
             """, LAST_NAME);
         var plan = plan(query, null);
-
-        var filter = filterQueryForTransportNodes(TransportVersion.current(), plan);
+        // test with an older version, so like list is not supported
+        var filter = filterQueryForTransportNodes(V_8_17_0, plan);
         assertNull(filter);
     }
 
@@ -342,12 +345,13 @@ public class FilterTests extends ESTestCase {
             """, LAST_NAME);
         var plan = plan(query, null);
 
-        SingleValueQuery.Builder filter = (SingleValueQuery.Builder) filterQueryForTransportNodes(null, plan);
+        PlanStreamWrapperQueryBuilder filterWrapper = (PlanStreamWrapperQueryBuilder) filterQueryForTransportNodes(null, plan);
+        SingleValueQuery.Builder filter = (SingleValueQuery.Builder) filterWrapper.next();
         assertEquals(LAST_NAME, filter.fieldName());
-        AutomatonQueryBuilder innerFilter = (AutomatonQueryBuilder) filter.next();
+        ExpressionQueryBuilder innerFilter = (ExpressionQueryBuilder) filter.next();
         assertEquals(LAST_NAME, innerFilter.fieldName());
         assertEquals("""
-            LIKE("a+", "b+"), caseInsensitive=false""", innerFilter.description());
+            last_name LIKE ("a+", "b+")""", innerFilter.getExpression().toString());
     }
 
     /**
@@ -378,7 +382,7 @@ public class FilterTests extends ESTestCase {
     }
 
     private PhysicalPlan plan(String query, QueryBuilder restFilter) {
-        var logical = logicalOptimizer.optimize(analyzer.analyze(parser.createStatement(query)));
+        var logical = logicalOptimizer.optimize(analyzer.analyze(parser.createStatement(query, EsqlTestUtils.TEST_CFG)));
         // System.out.println("Logical\n" + logical);
         var physical = mapper.map(logical);
         // System.out.println("physical\n" + physical);
@@ -397,7 +401,7 @@ public class FilterTests extends ESTestCase {
     }
 
     private QueryBuilder filterQueryForTransportNodes(TransportVersion minTransportVersion, PhysicalPlan plan) {
-        return PlannerUtils.detectFilter(minTransportVersion, plan, Set.of(EMP_NO, LAST_NAME)::contains);
+        return PlannerUtils.detectFilter(new EsqlFlags(true), null, minTransportVersion, plan, Set.of(EMP_NO, LAST_NAME)::contains);
     }
 
     @Override

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

@@ -201,7 +201,7 @@ public class ClusterRequestTests extends AbstractWireSerializingTestCase<Cluster
             ),
             TEST_VERIFIER
         );
-        return logicalOptimizer.optimize(analyzer.analyze(new EsqlParser().createStatement(query)));
+        return logicalOptimizer.optimize(analyzer.analyze(new EsqlParser().createStatement(query, EsqlTestUtils.TEST_CFG)));
     }
 
     @Override

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

@@ -302,7 +302,7 @@ public class DataNodeRequestSerializationTests extends AbstractWireSerializingTe
             ),
             TEST_VERIFIER
         );
-        return logicalOptimizer.optimize(analyzer.analyze(new EsqlParser().createStatement(query)));
+        return logicalOptimizer.optimize(analyzer.analyze(new EsqlParser().createStatement(query, EsqlTestUtils.TEST_CFG)));
     }
 
     static PhysicalPlan mapAndMaybeOptimize(LogicalPlan logicalPlan) {

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/session/IndexResolverFieldNamesTests.java

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.esql.session;
 
 import org.elasticsearch.Build;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.esql.EsqlTestUtils;
 import org.elasticsearch.xpack.esql.action.EsqlCapabilities;
 import org.elasticsearch.xpack.esql.parser.EsqlParser;
 import org.elasticsearch.xpack.esql.parser.ParsingException;
@@ -1600,7 +1601,7 @@ public class IndexResolverFieldNamesTests extends ESTestCase {
     public void testMetrics() {
         var query = "TS k8s | STATS bytes=sum(rate(network.total_bytes_in)), sum(rate(network.total_cost)) BY cluster";
         if (Build.current().isSnapshot() == false) {
-            var e = expectThrows(ParsingException.class, () -> parser.createStatement(query));
+            var e = expectThrows(ParsingException.class, () -> parser.createStatement(query, EsqlTestUtils.TEST_CFG));
             assertThat(e.getMessage(), containsString("line 1:1: mismatched input 'TS' expecting {"));
             return;
         }
@@ -2148,7 +2149,8 @@ public class IndexResolverFieldNamesTests extends ESTestCase {
 
     private Set<String> fieldNames(String query, Set<String> enrichPolicyMatchFields) {
         var preAnalysisResult = new EsqlSession.PreAnalysisResult(null);
-        return EsqlSession.fieldNames(parser.createStatement(query), enrichPolicyMatchFields, preAnalysisResult).fieldNames();
+        return EsqlSession.fieldNames(parser.createStatement(query, EsqlTestUtils.TEST_CFG), enrichPolicyMatchFields, preAnalysisResult)
+            .fieldNames();
     }
 
     private void assertFieldNames(String query, Set<String> expected) {
@@ -2157,7 +2159,11 @@ public class IndexResolverFieldNamesTests extends ESTestCase {
     }
 
     private void assertFieldNames(String query, Set<String> expected, Set<String> wildCardIndices) {
-        var preAnalysisResult = EsqlSession.fieldNames(parser.createStatement(query), Set.of(), new EsqlSession.PreAnalysisResult(null));
+        var preAnalysisResult = EsqlSession.fieldNames(
+            parser.createStatement(query, EsqlTestUtils.TEST_CFG),
+            Set.of(),
+            new EsqlSession.PreAnalysisResult(null)
+        );
         assertThat("Query-wide field names", preAnalysisResult.fieldNames(), equalTo(expected));
         assertThat("Lookup Indices that expect wildcard lookups", preAnalysisResult.wildcardJoinIndices(), equalTo(wildCardIndices));
     }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/telemetry/VerifierMetricsTests.java

@@ -10,6 +10,7 @@ package org.elasticsearch.xpack.esql.telemetry;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.core.watcher.common.stats.Counters;
+import org.elasticsearch.xpack.esql.EsqlTestUtils;
 import org.elasticsearch.xpack.esql.analysis.Verifier;
 import org.elasticsearch.xpack.esql.expression.function.EsqlFunctionRegistry;
 import org.elasticsearch.xpack.esql.expression.function.FunctionDefinition;
@@ -554,7 +555,7 @@ public class VerifierMetricsTests extends ESTestCase {
             metrics = new Metrics(new EsqlFunctionRegistry());
             verifier = new Verifier(metrics, new XPackLicenseState(() -> 0L));
         }
-        analyzer(verifier).analyze(parser.createStatement(esql));
+        analyzer(verifier).analyze(parser.createStatement(esql, EsqlTestUtils.TEST_CFG));
 
         return metrics == null ? null : metrics.stats();
     }

x-pack/plugin/mapper-constant-keyword/src/main/java/org/elasticsearch/xpack/constantkeyword/mapper/ConstantKeywordFieldMapper.java

@@ -212,6 +212,11 @@ public class ConstantKeywordFieldMapper extends FieldMapper {
             return Regex.simpleMatch(pattern, value, caseInsensitive);
         }
 
+        @Override
+        public String getConstantFieldValue(SearchExecutionContext context) {
+            return value;
+        }
+
         @Override
         public Query existsQuery(SearchExecutionContext context) {
             return value != null ? new MatchAllDocsQuery() : new MatchNoDocsQuery();