more fixes

Author: smalyshev

.idea/runConfigurations/Debug_Elasticsearch__node_2_.xml

@@ -70,20 +70,20 @@ public void testLookupJoinAcrossClusters() throws IOException {
             assertCCSExecutionInfoDetails(executionInfo);
         }
 
-        // populateLookupIndex(LOCAL_CLUSTER, "values_lookup2", 5);
-        // populateLookupIndex(REMOTE_CLUSTER_1, "values_lookup2", 5);
+        populateLookupIndex(LOCAL_CLUSTER, "values_lookup2", 5);
+        populateLookupIndex(REMOTE_CLUSTER_1, "values_lookup2", 5);
         // FIXME: this currently does not work
-        // try (
-        // EsqlQueryResponse resp = runQuery(
-        // "FROM logs-*,c*:logs-* | EVAL lookup_key = v | LOOKUP JOIN values_lookup ON lookup_key | LOOKUP JOIN values_lookup2 ON
-        // lookup_tag",
-        // randomBoolean()
-        // )
-        // ) {
-        // List<List<Object>> values = getValuesList(resp);
-        // assertThat(values, hasSize(20));
-        //
-        // }
+        try (
+            EsqlQueryResponse resp = runQuery(
+                "FROM logs-*,c*:logs-* | EVAL lookup_key = v | LOOKUP JOIN values_lookup ON lookup_key "
+                    + "| LOOKUP JOIN values_lookup2 ON lookup_tag",
+                randomBoolean()
+            )
+        ) {
+            List<List<Object>> values = getValuesList(resp);
+            assertThat(values, hasSize(20));
+
+        }
     }
 
     public void testLookupJoinMissingRemoteIndex() throws IOException {

.idea/runConfigurations/Debug_Elasticsearch__node_3_.xml

@@ -83,6 +83,7 @@ public class Join extends BinaryPlan implements PostAnalysisVerificationAware, S
 
     private final JoinConfig config;
     private List<Attribute> lazyOutput;
+    private transient boolean isRemote = false;
 
     public Join(Source source, LogicalPlan left, LogicalPlan right, JoinConfig config) {
         super(source, left, right);
@@ -257,7 +258,10 @@ public boolean equals(Object obj) {
         }
 
         Join other = (Join) obj;
-        return config.equals(other.config) && Objects.equals(left(), other.left()) && Objects.equals(right(), other.right());
+        return config.equals(other.config)
+            && Objects.equals(left(), other.left())
+            && Objects.equals(right(), other.right())
+            && isRemote == other.isRemote;
     }
 
     @Override
@@ -295,4 +299,13 @@ private static boolean comparableTypes(Attribute left, Attribute right) {
         }
         return leftType.noText() == rightType.noText();
     }
+
+    public boolean isRemote() {
+        return isRemote;
+    }
+
+    public Join setRemote(boolean remote) {
+        isRemote = remote;
+        return this;
+    }
 }

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterLookupJoinIT.java

@@ -32,8 +32,6 @@
  */
 public class LookupJoin extends Join implements SurrogateLogicalPlan, PostAnalysisVerificationAware, TelemetryAware {
 
-    private boolean isRemote = false;
-
     public LookupJoin(Source source, LogicalPlan left, LogicalPlan right, List<Attribute> joinFields) {
         this(source, left, right, new UsingJoinType(LEFT, joinFields), emptyList(), emptyList(), emptyList());
     }
@@ -60,7 +58,7 @@ public LookupJoin(Source source, LogicalPlan left, LogicalPlan right, JoinConfig
     @Override
     public LogicalPlan surrogate() {
         // TODO: decide whether to introduce USING or just basic ON semantics - keep the ordering out for now
-        return new Join(source(), left(), right(), config());
+        return new Join(source(), left(), right(), config()).setRemote(isRemote());
     }
 
     @Override
@@ -90,7 +88,7 @@ public String telemetryLabel() {
     @Override
     public void postAnalysisVerification(Failures failures) {
         super.postAnalysisVerification(failures);
-        if (isRemote) {
+        if (isRemote()) {
             checkRemoteJoin(failures);
         }
     }
@@ -120,12 +118,4 @@ private void checkRemoteJoin(Failures failures) {
         }
     }
 
-    public boolean isRemote() {
-        return isRemote;
-    }
-
-    public LookupJoin setRemote(boolean remote) {
-        isRemote = remote;
-        return this;
-    }
 }

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

@@ -201,7 +201,7 @@ private PhysicalPlan mapUnary(UnaryPlan unary) {
         return MapperUtils.mapUnary(unary, mappedChild);
     }
 
-    private PhysicalPlan mapToFragmentExec(LogicalPlan logical, PhysicalPlan child) {
+    private PhysicalPlan mapToFragmentExec(Join logical, PhysicalPlan child) {
         Holder<Boolean> hasFragment = new Holder<>(false);
         Holder<Boolean> forceLocal = new Holder<>(false);
 
@@ -244,10 +244,17 @@ private PhysicalPlan mapToFragmentExec(LogicalPlan logical, PhysicalPlan child)
             if (f instanceof LookupJoinExec lj) {
                 return lj.right();
             }
+            if (f instanceof MergeExec) {
+                forceLocal.set(true);
+                return f;
+            }
             return f;
         });
 
         if (forceLocal.get()) {
+            if (logical.isRemote()) {
+                throw new EsqlIllegalArgumentException("Remote joins are not supported in this context");
+            }
             return null;
         }
 
@@ -279,7 +286,7 @@ private PhysicalPlan mapBinary(BinaryPlan bp) {
             }
 
             if (FRAGMENT_EXEC_HACK_ENABLED) {
-                var leftPlan = mapToFragmentExec(bp, left);
+                var leftPlan = mapToFragmentExec(join, left);
                 if (leftPlan != null) {
                     return leftPlan;
                 }

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

@@ -441,12 +441,23 @@ private PreAnalysisResult receiveLookupIndexResolution(
         IndexResolution newIndexResolution
     ) {
         EsqlCCSUtils.updateExecutionInfoWithUnavailableClusters(executionInfo, newIndexResolution.unavailableClusters());
-        if (newIndexResolution.isValid() == false
-            || executionInfo.getClusters().isEmpty()
-            || executionInfo.isCrossClusterSearch() == false) {
+        if (newIndexResolution.isValid() == false) {
             // If the index resolution is invalid, or we're not dealing with CCS, don't bother with the rest of the analysis
             return result.addLookupIndexResolution(index, newIndexResolution);
         }
+        if (executionInfo.getClusters().isEmpty() || executionInfo.isCrossClusterSearch() == false) {
+            // Local only case, still do some checks
+            if (newIndexResolution.get().indexNameWithModes().size() > 1) {
+                throw new VerificationException("multiple resolutions for lookup index [" + index + "] in local cluster");
+            }
+            var indexMode = newIndexResolution.get().indexNameWithModes().entrySet().iterator().next().getValue();
+            if (indexMode != IndexMode.LOOKUP) {
+                throw new VerificationException(
+                    "invalid [" + index + "] resolution in lookup mode to an index in [" + indexMode + "] mode"
+                );
+            }
+            return result.addLookupIndexResolution(index, newIndexResolution);
+        }
         // Collect resolved clusters from the index resolution, verify that each cluster has a single resolution for the lookup index
         Map<String, String> clustersWithResolvedIndices = new HashMap<>(newIndexResolution.resolvedIndices().size());
         newIndexResolution.get().indexNameWithModes().forEach((indexName, indexMode) -> {

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

@@ -2898,6 +2898,7 @@ public void testVerifierOnMissingReferences() {
         assertThat(e.getMessage(), containsString(" > 10[INTEGER]]] optimized incorrectly due to missing references [emp_no{f}#"));
     }
 
+    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/118531")
     public void testVerifierOnMissingReferencesWithBinaryPlans() throws Exception {
         assumeTrue("Requires LOOKUP JOIN", EsqlCapabilities.Cap.JOIN_LOOKUP_V12.isEnabled());
 
@@ -7784,7 +7785,7 @@ public void testLookupJoinFieldLoadingDropAllFields() throws Exception {
     }
 
     private void assertLookupJoinFieldNames(String query, TestDataSource data, List<Set<String>> expectedFieldNames) {
-        assertLookupJoinFieldNames(query, data, expectedFieldNames, false);
+        assertLookupJoinFieldNames(query, data, expectedFieldNames, true);
     }
 
     private void assertLookupJoinFieldNames(