ESQL: introduce a pre-mapping logical plan processing step

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/util/Holder.java

@@ -26,6 +26,19 @@ public void set(T value) {
         this.value = value;
     }
 
+    /**
+     * Sets a value in the holder, but only if none has already been set.
+     * @param value the new value to set.
+     * @return the previously held value, if any was set.
+     */
+    public T trySet(T value) {
+        T old = this.value;
+        if (old == null) {
+            this.value = value;
+        }
+        return old;
+    }
+
     public T get() {
         return value;
     }

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

@@ -11,6 +11,8 @@
 import org.apache.lucene.sandbox.document.HalfFloatPoint;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.breaker.NoopCircuitBreaker;
@@ -31,9 +33,11 @@
 import org.elasticsearch.geo.ShapeTestUtils;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.search.SearchService;
 import org.elasticsearch.tasks.TaskCancelledException;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.transport.RemoteTransportException;
+import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xcontent.json.JsonXContent;
 import org.elasticsearch.xpack.esql.action.EsqlQueryResponse;
 import org.elasticsearch.xpack.esql.analysis.EnrichResolution;
@@ -72,8 +76,8 @@
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalSupplier;
 import org.elasticsearch.xpack.esql.plugin.EsqlPlugin;
 import org.elasticsearch.xpack.esql.plugin.QueryPragmas;
+import org.elasticsearch.xpack.esql.plugin.TransportActionServices;
 import org.elasticsearch.xpack.esql.session.Configuration;
-import org.elasticsearch.xpack.esql.session.QueryBuilderResolver;
 import org.elasticsearch.xpack.esql.stats.SearchStats;
 import org.elasticsearch.xpack.esql.telemetry.Metrics;
 import org.elasticsearch.xpack.versionfield.Version;
@@ -140,6 +144,7 @@
 import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.mockito.Mockito.mock;
 
 public final class EsqlTestUtils {
 
@@ -360,7 +365,14 @@ public static LogicalOptimizerContext unboundLogicalOptimizerContext() {
 
     public static final Verifier TEST_VERIFIER = new Verifier(new Metrics(new EsqlFunctionRegistry()), new XPackLicenseState(() -> 0L));
 
-    public static final QueryBuilderResolver MOCK_QUERY_BUILDER_RESOLVER = new MockQueryBuilderResolver();
+    public static final TransportActionServices MOCK_TRANSPORT_ACTION_SERVICES = new TransportActionServices(
+        mock(TransportService.class),
+        mock(SearchService.class),
+        null,
+        mock(ClusterService.class),
+        mock(IndexNameExpressionResolver.class),
+        null
+    );
 
     private EsqlTestUtils() {}
 

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

@@ -21,10 +21,10 @@
 import org.elasticsearch.xpack.esql.optimizer.LogicalOptimizerContext;
 import org.elasticsearch.xpack.esql.optimizer.LogicalPlanOptimizer;
 import org.elasticsearch.xpack.esql.planner.mapper.Mapper;
+import org.elasticsearch.xpack.esql.plugin.TransportActionServices;
 import org.elasticsearch.xpack.esql.session.Configuration;
 import org.elasticsearch.xpack.esql.session.EsqlSession;
 import org.elasticsearch.xpack.esql.session.IndexResolver;
-import org.elasticsearch.xpack.esql.session.QueryBuilderResolver;
 import org.elasticsearch.xpack.esql.session.Result;
 import org.elasticsearch.xpack.esql.telemetry.Metrics;
 import org.elasticsearch.xpack.esql.telemetry.PlanTelemetry;
@@ -62,7 +62,7 @@ public void esql(
         EsqlExecutionInfo executionInfo,
         IndicesExpressionGrouper indicesExpressionGrouper,
         EsqlSession.PlanRunner planRunner,
-        QueryBuilderResolver queryBuilderResolver,
+        TransportActionServices services,
         ActionListener<Result> listener
     ) {
         final PlanTelemetry planTelemetry = new PlanTelemetry(functionRegistry);
@@ -78,7 +78,7 @@ public void esql(
             verifier,
             planTelemetry,
             indicesExpressionGrouper,
-            queryBuilderResolver
+            services
         );
         QueryMetric clientId = QueryMetric.fromString("rest");
         metrics.total(clientId);

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/FullTextFunction.java

@@ -7,7 +7,7 @@
 
 package org.elasticsearch.xpack.esql.expression.function.fulltext;
 
-import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.compute.lucene.LuceneQueryExpressionEvaluator;
 import org.elasticsearch.compute.lucene.LuceneQueryExpressionEvaluator.ShardConfig;
 import org.elasticsearch.compute.operator.EvalOperator;
@@ -110,11 +110,7 @@ public Expression query() {
      */
     public Object queryAsObject() {
         Object queryAsObject = query().fold(FoldContext.small() /* TODO remove me */);
-        if (queryAsObject instanceof BytesRef bytesRef) {
-            return bytesRef.utf8ToString();
-        }
-
-        return queryAsObject;
+        return BytesRefs.toString(queryAsObject);
     }
 
     @Override

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/QueryBuilderResolver.java

@@ -0,0 +1,97 @@
+/*
+ * 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.expression.function.fulltext;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ResolvedIndices;
+import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.index.query.QueryRewriteContext;
+import org.elasticsearch.index.query.Rewriteable;
+import org.elasticsearch.xpack.esql.core.util.Holder;
+import org.elasticsearch.xpack.esql.plan.logical.EsRelation;
+import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
+import org.elasticsearch.xpack.esql.planner.TranslatorHandler;
+import org.elasticsearch.xpack.esql.plugin.TransportActionServices;
+import org.elasticsearch.xpack.esql.session.IndexResolver;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Some {@link FullTextFunction} implementations such as {@link org.elasticsearch.xpack.esql.expression.function.fulltext.Match}
+ * will be translated to a {@link QueryBuilder} that require a rewrite phase on the coordinator.
+ * {@link QueryBuilderResolver#resolveQueryBuilders(LogicalPlan, TransportActionServices, ActionListener)} will rewrite the plan by
+ * replacing {@link FullTextFunction} expression with new ones that hold rewritten {@link QueryBuilder}s.
+ */
+public final class QueryBuilderResolver {
+
+    public static final QueryBuilderResolver INSTANCE = new QueryBuilderResolver();
+
+    private QueryBuilderResolver() {}
+
+    public void resolveQueryBuilders(LogicalPlan plan, TransportActionServices services, ActionListener<LogicalPlan> listener) {
+        var hasFullTextFunctions = plan.anyMatch(p -> {
+            Holder<Boolean> hasFullTextFunction = new Holder<>(false);
+            p.forEachExpression(FullTextFunction.class, unused -> hasFullTextFunction.set(true));
+            return hasFullTextFunction.get();
+        });
+        if (hasFullTextFunctions) {
+            Rewriteable.rewriteAndFetch(
+                new FullTextFunctionsRewritable(plan),
+                queryRewriteContext(services, indexNames(plan)),
+                listener.delegateFailureAndWrap((l, r) -> l.onResponse(r.plan))
+            );
+        } else {
+            listener.onResponse(plan);
+        }
+    }
+
+    private static QueryRewriteContext queryRewriteContext(TransportActionServices services, Set<String> indexNames) {
+        ResolvedIndices resolvedIndices = ResolvedIndices.resolveWithIndexNamesAndOptions(
+            indexNames.toArray(String[]::new),
+            IndexResolver.FIELD_CAPS_INDICES_OPTIONS,
+            services.clusterService().state(),
+            services.indexNameExpressionResolver(),
+            services.transportService().getRemoteClusterService(),
+            System.currentTimeMillis()
+        );
+
+        return services.searchService().getRewriteContext(System::currentTimeMillis, resolvedIndices, null);
+    }
+
+    public Set<String> indexNames(LogicalPlan plan) {
+        Set<String> indexNames = new HashSet<>();
+        plan.forEachDown(EsRelation.class, esRelation -> indexNames.addAll(esRelation.concreteIndices()));
+        return indexNames;
+    }
+
+    private record FullTextFunctionsRewritable(LogicalPlan plan) implements Rewriteable<QueryBuilderResolver.FullTextFunctionsRewritable> {
+        @Override
+        public FullTextFunctionsRewritable rewrite(QueryRewriteContext ctx) throws IOException {
+            Holder<IOException> exceptionHolder = new Holder<>();
+            Holder<Boolean> updated = new Holder<>(false);
+            LogicalPlan newPlan = plan.transformExpressionsDown(FullTextFunction.class, f -> {
+                QueryBuilder builder = f.queryBuilder(), initial = builder;
+                builder = builder == null ? f.asQuery(TranslatorHandler.TRANSLATOR_HANDLER).asBuilder() : builder;
+                try {
+                    builder = builder.rewrite(ctx);
+                } catch (IOException e) {
+                    exceptionHolder.trySet(e);
+                }
+                var rewritten = builder != initial;
+                updated.set(updated.get() || rewritten);
+                return rewritten ? f.replaceQueryBuilder(builder) : f;
+            });
+            if (exceptionHolder.get() != null) {
+                throw exceptionHolder.get();
+            }
+            return updated.get() ? new FullTextFunctionsRewritable(newPlan) : this;
+        }
+    }
+}

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

@@ -0,0 +1,37 @@
+/*
+ * 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.planner.premapper;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.xpack.esql.expression.function.fulltext.QueryBuilderResolver;
+import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
+import org.elasticsearch.xpack.esql.plugin.TransportActionServices;
+
+/**
+ * The class is responsible for invoking any premapping steps that need to be applied to the logical plan,
+ * before this is being mapped to a physical one.
+ */
+public class PreMapper {
+
+    private final TransportActionServices services;
+
+    public PreMapper(TransportActionServices services) {
+        this.services = services;
+    }
+
+    /**
+     * Invokes any premapping steps that need to be applied to the logical plan, before this is being mapped to a physical one.
+     */
+    public void preMapper(LogicalPlan plan, ActionListener<LogicalPlan> listener) {
+        queryRewrite(plan, listener);
+    }
+
+    private void queryRewrite(LogicalPlan plan, ActionListener<LogicalPlan> listener) {
+        QueryBuilderResolver.INSTANCE.resolveQueryBuilders(plan, services, listener);
+    }
+}

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

@@ -0,0 +1,24 @@
+/*
+ * 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.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.compute.operator.exchange.ExchangeService;
+import org.elasticsearch.search.SearchService;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.usage.UsageService;
+
+public record TransportActionServices(
+    TransportService transportService,
+    SearchService searchService,
+    ExchangeService exchangeService,
+    ClusterService clusterService,
+    IndexNameExpressionResolver indexNameExpressionResolver,
+    UsageService usageService
+) {}

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

@@ -53,7 +53,6 @@
 import org.elasticsearch.xpack.esql.execution.PlanExecutor;
 import org.elasticsearch.xpack.esql.session.Configuration;
 import org.elasticsearch.xpack.esql.session.EsqlSession.PlanRunner;
-import org.elasticsearch.xpack.esql.session.QueryBuilderResolver;
 import org.elasticsearch.xpack.esql.session.Result;
 
 import java.io.IOException;
@@ -81,8 +80,8 @@ public class TransportEsqlQueryAction extends HandledTransportAction<EsqlQueryRe
     private final LookupFromIndexService lookupFromIndexService;
     private final AsyncTaskManagementService<EsqlQueryRequest, EsqlQueryResponse, EsqlQueryTask> asyncTaskManagementService;
     private final RemoteClusterService remoteClusterService;
-    private final QueryBuilderResolver queryBuilderResolver;
     private final UsageService usageService;
+    private final TransportActionServices services;
     // Listeners for active async queries, key being the async task execution ID
     private final Map<String, EsqlQueryListener> asyncListeners = ConcurrentCollections.newConcurrentMap();
 
@@ -153,8 +152,16 @@ public TransportEsqlQueryAction(
             bigArrays
         );
         this.remoteClusterService = transportService.getRemoteClusterService();
-        this.queryBuilderResolver = new QueryBuilderResolver(searchService, clusterService, transportService, indexNameExpressionResolver);
         this.usageService = usageService;
+
+        this.services = new TransportActionServices(
+            transportService,
+            searchService,
+            exchangeService,
+            clusterService,
+            indexNameExpressionResolver,
+            usageService
+        );
     }
 
     @Override
@@ -258,7 +265,7 @@ private void innerExecute(Task task, EsqlQueryRequest request, ActionListener<Es
             executionInfo,
             remoteClusterService,
             planRunner,
-            queryBuilderResolver,
+            services,
             ActionListener.wrap(result -> {
                 recordCCSTelemetry(task, executionInfo, request, null);
                 listener.onResponse(toResponse(task, request, configuration, result));