Refactor mapper, allow SORT and ban LIMIT pre-remote join

Author: smalyshev

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

@@ -119,11 +119,11 @@ public MultiClusterSpecIT(
         "StatsAndLookupMessageFromIndex",
         "MvJoinKeyOnTheLookupIndexAfterStats",
         "MvJoinKeyOnFromAfterStats",
-        // Lookup join after SORT is not supported in CCS yet
-        "SortBeforeAndAfterJoin",
-        "SortEvalBeforeLookup",
-        "NullifiedJoinKeyToPurgeTheJoin",
-        "SortBeforeAndAfterMultipleJoinAndMvExpand"
+        // Lookup join after SORT+LIMIT is not supported in CCS yet
+        "NullifiedJoinKeyToPurgeTheJoin"
+    // "SortBeforeAndAfterJoin",
+    // "SortEvalBeforeLookup",
+    // "SortBeforeAndAfterMultipleJoinAndMvExpand"
     );
 
     @Override
@@ -148,7 +148,7 @@ protected void shouldSkipTest(String testName) throws IOException {
         assumeFalse("INLINESTATS not yet supported in CCS", testCase.requiredCapabilities.contains(INLINESTATS_V7.capabilityName()));
         if (testCase.requiredCapabilities.contains(JOIN_LOOKUP_V12.capabilityName())) {
             assumeTrue("LOKUP JOIN not supported", supportsIndexModeLookup());
-            assumeTrue("LOOKUP JOIN not yet supported in CCS", Clusters.localClusterVersion().onOrAfter(Version.fromString("9.1.0")));
+            assumeTrue("LOOKUP JOIN not yet supported in CCS", Clusters.localClusterVersion().onOrAfter(Version.fromString("9.2.0")));
         }
         // Unmapped fields require a coorect capability response from every cluster, which isn't currently implemented.
         assumeFalse("UNMAPPED FIELDS not yet supported in CCS", testCase.requiredCapabilities.contains(UNMAPPED_FIELDS.capabilityName()));

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

@@ -150,8 +150,7 @@ private static boolean checkVersion(org.elasticsearch.Version version) {
     }
 
     public void testIndicesDontExistWithRemotePattern() throws IOException {
-        // TODO: add 8.19 if this is merged to 8.x
-        assumeTrue("Only works with remote LOOKUP JOIN support", Clusters.localClusterVersion().onOrAfter(Version.fromString("9.1.0")));
+        assumeTrue("Only works with remote LOOKUP JOIN support", Clusters.localClusterVersion().onOrAfter(Version.fromString("9.2.0")));
 
         int docsTest1 = randomIntBetween(1, 5);
         indexTimestampData(docsTest1, "test1", "2024-11-26", "id1");

x-pack/plugin/esql/qa/testFixtures/src/main/resources/lookup-join.csv-spec

@@ -331,7 +331,7 @@ emp_no:integer
 10001
 ;
 
-dropAllFieldsUsedInLookupOnTheCoordinator
+dropAllFieldsUsedInLookupOnMaybeTheCoordinator
 required_capability: join_lookup_v12
 
 FROM employees
@@ -367,7 +367,7 @@ language_code:integer | language_name:keyword       | country:text
 1                     | English                     | null
 ;
 
-nonUniqueRightKeyOnTheCoordinatorLateLimit
+nonUniqueRightKeyOnMaybeTheCoordinatorLateLimit
 required_capability: join_lookup_v12
 required_capability: join_lookup_fix_limit_pushdown
 
@@ -714,7 +714,7 @@ emp_no:integer | language_code:integer | language_name:keyword
 10003          | 4                     | German
 ;
 
-filterOnTheDataNodeThenFilterOnTheCoordinator
+filterOnTheDataNodeThenFilterOnMaybeTheCoordinator
 required_capability: join_lookup_v12
 
 FROM employees

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

@@ -15,6 +15,7 @@
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.plan.logical.Aggregate;
 import org.elasticsearch.xpack.esql.plan.logical.Enrich;
+import org.elasticsearch.xpack.esql.plan.logical.Limit;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.esql.plan.logical.OrderBy;
 import org.elasticsearch.xpack.esql.plan.logical.SurrogateLogicalPlan;
@@ -98,15 +99,15 @@ public void postAnalysisVerification(Failures failures) {
     private void checkRemoteJoin(Failures failures) {
         boolean[] agg = { false };
         boolean[] enrichCoord = { false };
-        boolean[] sort = { false };
+        boolean[] limit = { false };
 
         this.forEachUp(UnaryPlan.class, u -> {
             if (u instanceof Aggregate) {
                 agg[0] = true;
             } else if (u instanceof Enrich enrich && enrich.mode() == Enrich.Mode.COORDINATOR) {
                 enrichCoord[0] = true;
-            } else if (u instanceof OrderBy) {
-                sort[0] = true;
+            } else if (u instanceof Limit) {
+                limit[0] = true;
             }
         });
         if (agg[0]) {
@@ -115,8 +116,8 @@ private void checkRemoteJoin(Failures failures) {
         if (enrichCoord[0]) {
             failures.add(fail(this, "LOOKUP JOIN with remote indices can't be executed after ENRICH with coordinator policy"));
         }
-        if (sort[0]) {
-            failures.add(fail(this, "LOOKUP JOIN with remote indices can't be executed after SORT"));
+        if (limit[0]) {
+            failures.add(fail(this, "LOOKUP JOIN with remote indices can't be executed after LIMIT"));
         }
     }
 

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

@@ -230,12 +230,13 @@ private PhysicalPlan mapToFragmentExec(Join logical, PhysicalPlan child) {
                 }
             }
             if (f instanceof UnaryExec unaryExec) {
-                if (f instanceof AggregateExec || f instanceof TopNExec) {
+                if (f instanceof AggregateExec) {
                     // We can't make a fragment here...
+                    assert logical.isRemote() == false : "Unexpected Aggregate inside remote join";
                     forceLocal.set(true);
                     return f;
                 }
-                if (f instanceof LimitExec || f instanceof ExchangeExec) {
+                if (f instanceof LimitExec || f instanceof ExchangeExec || f instanceof TopNExec) {
                     return f;
                 } else {
                     return unaryExec.child();