Cleanups

Author: smalyshev

docs/changelog/129013.yaml

@@ -1,5 +1,12 @@
 pr: 129013
-summary: "WIP: Remote lookup join implementation"
+summary: "Add remote index support to LOOKUP JOIN"
 area: ES|QL
-type: enhancement
-issues: []
+type: feature
+issues: [ ]
+highlight:
+  title: Add remote index support to LOOKUP JOIN
+  body: |-
+    Queries containing LOOKUP JOIN now can be preformed on cross-cluster indices, for example:
+    ```
+    FROM logs-*, remote:logs-* | LOOKUP JOIN clients on ip | SORT timestamp | LIMIT 100
+    ```

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

@@ -37,6 +37,10 @@ public LookupJoin(Source source, LogicalPlan left, LogicalPlan right, List<Attri
         this(source, left, right, new UsingJoinType(LEFT, joinFields), emptyList(), emptyList(), emptyList(), isRemote);
     }
 
+    public LookupJoin(Source source, LogicalPlan left, LogicalPlan right, List<Attribute> joinFields) {
+        this(source, left, right, new UsingJoinType(LEFT, joinFields), emptyList(), emptyList(), emptyList(), false);
+    }
+
     public LookupJoin(
         Source source,
         LogicalPlan left,
@@ -50,6 +54,10 @@ public LookupJoin(
         this(source, left, right, new JoinConfig(type, joinFields, leftFields, rightFields), isRemote);
     }
 
+    public LookupJoin(Source source, LogicalPlan left, LogicalPlan right, JoinConfig joinConfig) {
+        this(source, left, right, joinConfig, false);
+    }
+
     public LookupJoin(Source source, LogicalPlan left, LogicalPlan right, JoinConfig joinConfig, boolean isRemote) {
         super(source, left, right, joinConfig, isRemote);
     }

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

@@ -736,6 +736,9 @@ private PhysicalOperation planLookupJoin(LookupJoinExec join, LocalExecutionPlan
             throw new IllegalArgumentException("can't plan [" + join + "]");
         }
 
+        // After enabling remote joins, we can have one of the two situations here:
+        // 1. We've just got one entry - this should be the one relevant to the join, and it should be for this cluster
+        // 2. We have got multiple entries - this means each cluster has its own one, and we should extract one relevant for this cluster
         Map.Entry<String, IndexMode> entry;
         if (localSourceExec.indexNameWithModes().size() == 1) {
             entry = localSourceExec.indexNameWithModes().entrySet().iterator().next();
@@ -756,6 +759,7 @@ private PhysicalOperation planLookupJoin(LookupJoinExec join, LocalExecutionPlan
             throw new IllegalArgumentException("can't plan [" + join + "], found index with mode [" + entry.getValue() + "]");
         }
         String[] indexSplit = RemoteClusterAware.splitIndexName(entry.getKey());
+        // No prefix is ok, prefix with this cluster is ok, something else is not
         if (indexSplit[0] != null && clusterAlias.equals(indexSplit[0]) == false) {
             throw new IllegalArgumentException(
                 "can't plan [" + join + "]: no matching index found " + EsqlCCSUtils.inClusterName(clusterAlias)

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

@@ -436,6 +436,7 @@ private void preAnalyzeLookupIndex(
         if (executionInfo.getClusters().isEmpty()) {
             patternWithRemotes = localPattern;
         } else {
+            // convert index -> cluster1:index,cluster2:index, etc.for each running cluster
             patternWithRemotes = executionInfo.getClusterStates(EsqlExecutionInfo.Cluster.Status.RUNNING)
                 .map(c -> RemoteClusterAware.buildRemoteIndexName(c.getClusterAlias(), localPattern))
                 .collect(Collectors.joining(","));
@@ -462,6 +463,10 @@ private void skipClusterOrError(String clusterAlias, EsqlExecutionInfo execution
         }
     }
 
+    /**
+     * Receive and process lookup index resolutions from resolveAsMergedMapping.
+     * This processes the lookup index data for a single index, updates and returns the {@link PreAnalysisResult} result
+     */
     private PreAnalysisResult receiveLookupIndexResolution(
         PreAnalysisResult result,
         String index,
@@ -474,9 +479,9 @@ private PreAnalysisResult receiveLookupIndexResolution(
             return result.addLookupIndexResolution(index, newIndexResolution);
         }
         if (executionInfo.getClusters().isEmpty() || executionInfo.isCrossClusterSearch() == false) {
-            // Local only case, still do some checks
+            // 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 + "] in local cluster");
+                throw new VerificationException("multiple resolutions for lookup index [" + index + "]");
             }
             var indexMode = newIndexResolution.get().indexNameWithModes().entrySet().iterator().next().getValue();
             if (indexMode != IndexMode.LOOKUP) {
@@ -495,7 +500,12 @@ private PreAnalysisResult receiveLookupIndexResolution(
                 skipClusterOrError(
                     clusterAlias,
                     executionInfo,
-                    "invalid [" + indexName + "] resolution in lookup mode to an index in [" + indexMode + "] mode"
+                    "invalid ["
+                        + indexName
+                        + "] resolution in lookup mode to an index in ["
+                        + indexMode
+                        + "] mode "
+                        + EsqlCCSUtils.inClusterName(clusterAlias)
                 );
             }
             // Each cluster should have only one resolution for the lookup index
@@ -526,6 +536,7 @@ private PreAnalysisResult receiveLookupIndexResolution(
         });
 
         // If all indices resolve to the same name, we can use that for BWC
+        // Older clusters only can handle one name in LOOKUP JOIN
         var indexNames = clustersWithResolvedIndices.values()
             .stream()
             .map(n -> RemoteClusterAware.splitIndexName(n)[1])
@@ -548,6 +559,12 @@ private PreAnalysisResult receiveLookupIndexResolution(
 
     }
 
+    /**
+     * Older clusters can only handle one name in LOCAL JOIN - verify that all the remotes involved
+     * are recent enough to be able to handle multiple indices.
+     * This is only checked if there are actually multiple indices, which happens when remotes have a different
+     * concrete indices aliased to the same index name.
+     */
     private void validateRemoteVersions(EsqlExecutionInfo executionInfo) {
         Stream<EsqlExecutionInfo.Cluster> clusters = executionInfo.getClusterStates(EsqlExecutionInfo.Cluster.Status.RUNNING);
         clusters.forEach(cluster -> {
@@ -559,9 +576,11 @@ private void validateRemoteVersions(EsqlExecutionInfo executionInfo) {
                     skipClusterOrError(
                         clusterAlias,
                         executionInfo,
-                        "remote cluster version ["
+                        "remote cluster ["
+                            + clusterAlias
+                            + "] has version ["
                             + connection.getTransportVersion()
-                            + "] does not support multiple indices in LOOKUP JOIN, skipping"
+                            + "] that does not support multiple indices in LOOKUP JOIN, skipping"
                     );
                 }
             }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java

@@ -1244,7 +1244,7 @@ public void testPushdownLimitsPastLeftJoin() {
         var joinConfig = new JoinConfig(JoinTypes.LEFT, List.of(), List.of(), List.of());
         var join = switch (randomIntBetween(0, 2)) {
             case 0 -> new Join(EMPTY, leftChild, rightChild, joinConfig);
-            case 1 -> new LookupJoin(EMPTY, leftChild, rightChild, joinConfig, false);
+            case 1 -> new LookupJoin(EMPTY, leftChild, rightChild, joinConfig);
             case 2 -> new InlineJoin(EMPTY, leftChild, rightChild, joinConfig);
             default -> throw new IllegalArgumentException();
         };

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/PhysicalPlanOptimizerTests.java

@@ -2898,7 +2898,6 @@ 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());
 

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/logical/CommandLicenseTests.java

@@ -164,7 +164,7 @@ private static LogicalPlan createInstance(Class<? extends LogicalPlan> clazz, Lo
                 return new Sample(source, null, child);
             }
             case "LookupJoin" -> {
-                return new LookupJoin(source, child, child, List.of(), false);
+                return new LookupJoin(source, child, child, List.of());
             }
             case "Limit" -> {
                 return new Limit(source, null, child);