Fix behavior for _index LIKE for ESQL (elastic#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 elastic#129511

(cherry picked from commit 8c4eaf9)

# Conflicts:
#	benchmarks/src/main/java/org/elasticsearch/benchmark/_nightly/esql/QueryPlanningBenchmark.java
#	server/src/main/java/org/elasticsearch/TransportVersions.java
#	server/src/main/java/org/elasticsearch/index/mapper/KeywordFieldMapper.java
#	server/src/main/java/org/elasticsearch/index/query/WildcardQueryBuilder.java
#	x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java
#	x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java
#	x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/DataNodeComputeHandler.java
#	x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/EsqlPlugin.java
#	x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/CsvTests.java
#	x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/AbstractLogicalPlanOptimizerTests.java
#	x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LocalPhysicalPlanOptimizerTests.java
#	x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java
#	x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PropagateInlineEvalsTests.java
#	x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/AbstractStatementParserTests.java
#	x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/session/IndexResolverFieldNamesTests.java

Author: julian-elastic

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

@@ -256,6 +256,7 @@ static TransportVersion def(int id) {
     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 = def(9_119_0_00);
 
     /*
      * 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.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 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 final Query wildcardQuery(String value, boolean caseInsensitive, QueryRew
         }
     }
 
+    /**
+     * 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 java.util.Collections;
 import java.util.List;
+import java.util.Locale;
 
 public class IndexFieldMapper extends MetadataFieldMapper {
 
@@ -102,6 +107,38 @@ public StoredFieldsSpec storedFieldsSpec() {
             };
         }
 
+        @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

@@ -60,6 +60,11 @@ protected boolean matches(String pattern, boolean caseInsensitive, QueryRewriteC
             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

@@ -23,13 +23,15 @@
 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;
 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.MinimizationOperations;
@@ -50,6 +52,7 @@
 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;
@@ -81,6 +84,7 @@
 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;
@@ -930,6 +934,17 @@ public boolean hasScriptValues() {
         public boolean hasNormalizer() {
             return normalizer != Lucene.KEYWORD_ANALYZER;
         }
+        }
+
+        @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

@@ -29,6 +29,8 @@
 import org.apache.lucene.search.TermInSetQuery;
 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;
@@ -58,6 +60,7 @@
 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;
 
@@ -333,6 +336,19 @@ public final Query wildcardQuery(String value, @Nullable MultiTermQuery.RewriteM
         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,
@@ -374,6 +390,23 @@ public Query regexpQuery(
         );
     }
 
+    /**
+     * 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());