Review feedback

Author: smalyshev

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

@@ -326,9 +326,9 @@ static TransportVersion def(int id) {
     public static final TransportVersion ML_INFERENCE_COHERE_API_VERSION = def(9_110_0_00);
     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 LOOKUP_JOIN_MANY_INDICES = def(9_113_0_00);
 
     public static final TransportVersion ESQL_SERIALIZE_TIMESERIES_FIELD_TYPE = def(9_113_0_00);
+    public static final TransportVersion LOOKUP_JOIN_CCS = def(9_114_0_00);
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/PipelineBreaker.java

@@ -7,4 +7,11 @@
 
 package org.elasticsearch.xpack.esql.plan.logical;
 
+import org.elasticsearch.xpack.esql.plan.physical.FragmentExec;
+
+/**
+ * A {@link LogicalPlan} that cannot be run only on the data nodes, resp. requires to be at least partially run on the coordinator.
+ * When mapping to a physical plan, the first pipeline breaker will give rise to a {@link FragmentExec}
+ * that contains the {@link LogicalPlan} that data nodes will execute.
+ */
 public interface PipelineBreaker {}

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

@@ -39,6 +39,7 @@
 import org.elasticsearch.xpack.esql.plan.QueryPlan;
 import org.elasticsearch.xpack.esql.plan.logical.EsRelation;
 import org.elasticsearch.xpack.esql.plan.logical.Filter;
+import org.elasticsearch.xpack.esql.plan.logical.PipelineBreaker;
 import org.elasticsearch.xpack.esql.plan.physical.AggregateExec;
 import org.elasticsearch.xpack.esql.plan.physical.EsSourceExec;
 import org.elasticsearch.xpack.esql.plan.physical.EstimatesRowSize;
@@ -49,7 +50,6 @@
 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.session.Configuration;
 import org.elasticsearch.xpack.esql.stats.SearchContextStats;
 import org.elasticsearch.xpack.esql.stats.SearchStats;
@@ -117,7 +117,7 @@ public static PhysicalPlan reductionPlan(PhysicalPlan plan) {
         }
         final FragmentExec fragment = (FragmentExec) fragments.getFirst();
 
-        final var pipelineBreakers = fragment.fragment().collectFirstChildren(Mapper::isPipelineBreaker);
+        final var pipelineBreakers = fragment.fragment().collectFirstChildren(p -> p instanceof PipelineBreaker);
         if (pipelineBreakers.isEmpty()) {
             return null;
         }

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

@@ -137,7 +137,7 @@ private PhysicalPlan mapUnary(UnaryPlan unary) {
                 return MapperUtils.mapUnary(unary, mappedChild);
             }
             // in case of a fragment, push to it any current streaming operator
-            if (isPipelineBreaker(unary) == false) {
+            if (unary instanceof PipelineBreaker == false) {
                 return new FragmentExec(unary);
             }
         }
@@ -253,10 +253,6 @@ private PhysicalPlan mapFork(Fork fork) {
         return new MergeExec(fork.source(), fork.children().stream().map(child -> map(child)).toList(), fork.output());
     }
 
-    public static boolean isPipelineBreaker(LogicalPlan p) {
-        return p instanceof PipelineBreaker;
-    }
-
     private PhysicalPlan addExchangeForFragment(LogicalPlan logical, PhysicalPlan child) {
         // in case of fragment, preserve the streaming operator (order-by, limit or topN) for local replanning
         // no need to do it for an aggregate since it gets split

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

@@ -481,12 +481,20 @@ private PreAnalysisResult receiveLookupIndexResolution(
         if (executionInfo.getClusters().isEmpty() || executionInfo.isCrossClusterSearch() == false) {
             // Local only case, still do some checks, since we moved analysis checks here
             if (newIndexResolution.get().indexNameWithModes().size() > 1) {
-                throw new VerificationException("multiple resolutions for lookup index [" + index + "]");
+                throw new VerificationException(
+                    "Lookup Join requires a single lookup mode index; [" + index + "] resolves to multiple indices"
+                );
             }
-            var indexMode = newIndexResolution.get().indexNameWithModes().entrySet().iterator().next().getValue();
-            if (indexMode != IndexMode.LOOKUP) {
+            var indexModeEntry = newIndexResolution.get().indexNameWithModes().entrySet().iterator().next();
+            if (indexModeEntry.getValue() != IndexMode.LOOKUP) {
                 throw new VerificationException(
-                    "invalid [" + index + "] resolution in lookup mode to an index in [" + indexMode + "] mode"
+                    "Lookup Join requires a single lookup mode index; ["
+                        + index
+                        + "] resolves to ["
+                        + indexModeEntry.getKey()
+                        + "] in ["
+                        + indexModeEntry.getValue()
+                        + "] mode"
                 );
             }
             return result.addLookupIndexResolution(index, newIndexResolution);
@@ -500,20 +508,24 @@ private PreAnalysisResult receiveLookupIndexResolution(
                 skipClusterOrError(
                     clusterAlias,
                     executionInfo,
-                    "invalid ["
+                    "Lookup Join requires a single lookup mode index; ["
+                        + index
+                        + "] resolves to ["
                         + indexName
-                        + "] resolution in lookup mode to an index in ["
+                        + "] in ["
                         + indexMode
-                        + "] mode "
-                        + EsqlCCSUtils.inClusterName(clusterAlias)
+                        + "] mode"
                 );
             }
             // Each cluster should have only one resolution for the lookup index
             if (clustersWithResolvedIndices.containsKey(clusterAlias)) {
                 skipClusterOrError(
                     clusterAlias,
                     executionInfo,
-                    "multiple resolutions for lookup index [" + index + "] " + EsqlCCSUtils.inClusterName(clusterAlias)
+                    "Lookup Join requires a single lookup mode index; ["
+                        + index
+                        + "] resolves to multiple indices "
+                        + EsqlCCSUtils.inClusterName(clusterAlias)
                 );
             } else {
                 clustersWithResolvedIndices.put(clusterAlias, indexName);
@@ -572,7 +584,7 @@ private void validateRemoteVersions(EsqlExecutionInfo executionInfo) {
             if (clusterAlias.equals(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY) == false) {
                 // No need to check local, obviously
                 var connection = remoteClusterService.getConnection(clusterAlias);
-                if (connection != null && connection.getTransportVersion().before(TransportVersions.LOOKUP_JOIN_MANY_INDICES)) {
+                if (connection != null && connection.getTransportVersion().before(TransportVersions.LOOKUP_JOIN_CCS)) {
                     skipClusterOrError(
                         clusterAlias,
                         executionInfo,

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/190_lookup_join.yml

@@ -160,7 +160,7 @@ fails with non-lookup index:
       catch: "bad_request"
 
   - match: { error.type: "verification_exception" }
-  - contains: { error.reason: "invalid [test] resolution in lookup mode to an index in [standard] mode" }
+  - contains: { error.reason: "Lookup Join requires a single lookup mode index; [test] resolves to [test] in [standard] mode" }
 
 ---
 pattern-multiple:

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/192_lookup_join_on_aliases.yml

@@ -201,7 +201,7 @@ fails when alias or pattern resolves to multiple:
       catch: "bad_request"
 
   - match: { error.type: "verification_exception" }
-  - contains: { error.reason: "multiple resolutions for lookup index [test-lookup-alias-pattern-multiple]" }
+  - contains: { error.reason: "Lookup Join requires a single lookup mode index; [test-lookup-alias-pattern-multiple] resolves to multiple indices" }
 
 ---
 alias-pattern-single: