Integrate with ExpressionQuery, fix merge errors

Author: julian-elastic

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

@@ -330,6 +330,8 @@ static TransportVersion def(int id) {
     public static final TransportVersion ESQL_SERIALIZE_TIMESERIES_FIELD_TYPE = def(9_114_0_00);
     public static final TransportVersion ML_INFERENCE_IBM_WATSONX_COMPLETION_ADDED = def(9_115_0_00);
     public static final TransportVersion ESQL_FIXED_INDEX_LIKE = def(9_116_0_00);
+    public static final TransportVersion ESQL_RLIKE_LIST = def(9_117_0_00);
+
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

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

@@ -1228,7 +1228,7 @@ public enum Cap {
         /**
          * Support for the RLIKE operator with a list of regexes.
          */
-        RLIKE_WITH_LIST_OF_PATTERNS
+        RLIKE_WITH_LIST_OF_PATTERNS,
 
         /**
          * FUSE command

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

@@ -7,23 +7,31 @@
 
 package org.elasticsearch.xpack.esql.expression.function.scalar.string.regex;
 
+import org.apache.lucene.search.MultiTermQuery;
+import org.apache.lucene.util.automaton.Automaton;
+import org.apache.lucene.util.automaton.CharacterRunAutomaton;
+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.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.RLikePattern;
 import org.elasticsearch.xpack.esql.core.expression.predicate.regex.RLikePatternList;
-import org.elasticsearch.xpack.esql.core.querydsl.query.EsqlAutomatonQuery;
 import org.elasticsearch.xpack.esql.core.querydsl.query.Query;
 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;
 
 public class RLikeList extends RegexMatch<RLikePatternList> {
@@ -33,6 +41,30 @@ public class RLikeList extends RegexMatch<RLikePatternList> {
         RLikeList::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 RLike, and shown to the users as `RLIKE` in the docs.
      */
@@ -75,39 +107,62 @@ public String getWriteableName() {
         return ENTRY.name;
     }
 
+    @Override
+    protected NodeInfo<RLikeList> info() {
+        return NodeInfo.create(this, RLikeList::new, field(), pattern(), caseInsensitive());
+    }
+
     @Override
     protected RLikeList replaceChild(Expression newChild) {
         return new RLikeList(source(), newChild, pattern(), caseInsensitive());
     }
 
+    /**
+     * Returns {@link Translatable#YES} if the field is pushable, otherwise {@link Translatable#NO}.
+     */
     @Override
     public Translatable translatable(LucenePushdownPredicates pushdownPredicates) {
-        return pushdownPredicates.isPushableAttribute(field()) ? Translatable.YES : Translatable.NO;
+        if (supportsPushdown(pushdownPredicates.minTransportVersion())) {
+            return pushdownPredicates.isPushableAttribute(field()) ? Translatable.YES : Translatable.NO;
+        } else {
+            // The ExpressionQuery we use isn't serializable to all nodes in the cluster.
+            return Translatable.NO;
+        }
     }
 
-    /**
-     * Returns a {@link Query} that matches the field against the provided patterns.
-     * For now, we only support a single pattern in the list for pushdown.
-     */
     @Override
     public Query asQuery(LucenePushdownPredicates pushdownPredicates, TranslatorHandler handler) {
         var field = field();
         LucenePushdownPredicates.checkIsPushableAttribute(field);
         return translateField(handler.nameOf(field instanceof FieldAttribute fa ? fa.exactAttribute() : field));
     }
 
-    private Query translateField(String targetFieldName) {
-        return new EsqlAutomatonQuery(source(), targetFieldName, pattern().createAutomaton(caseInsensitive()), getAutomatonDescription());
+    private boolean supportsPushdown(TransportVersion version) {
+        return version == null || version.onOrAfter(TransportVersions.ESQL_RLIKE_LIST);
     }
 
     @Override
-    protected NodeInfo<? extends Expression> info() {
-        return NodeInfo.create(this, RLikeList::new, field(), pattern(), caseInsensitive());
+    public org.apache.lucene.search.Query asLuceneQuery(
+        MappedFieldType fieldType,
+        MultiTermQuery.RewriteMethod constantScoreRewrite,
+        SearchExecutionContext context
+    ) {
+        return fieldType.automatonQuery(
+            automatonSupplier,
+            characterRunAutomatonSupplier,
+            constantScoreRewrite,
+            context,
+            getLuceneQueryDescription()
+        );
     }
 
-    private String getAutomatonDescription() {
+    private String getLuceneQueryDescription() {
         // we use the information used to create the automaton to describe the query here
         String patternDesc = pattern().patternList().stream().map(RLikePattern::pattern).collect(Collectors.joining("\", \""));
         return "RLIKE(\"" + patternDesc + "\"), caseInsensitive=" + caseInsensitive();
     }
+
+    private Query translateField(String targetFieldName) {
+        return new ExpressionQuery(source(), targetFieldName, this);
+    }
 }