Add more UTs and bugfixes

Author: julian-elastic

.idea/runConfigurations/Debug_Elasticsearch__node_2_.xml

@@ -118,6 +118,7 @@ public MultiClusterSpecIT(
         "LookupJoinOnTwoFieldsMultipleTimes",
         // Lookup join after LIMIT is not supported in CCS yet
         "LookupJoinAfterLimitAndRemoteEnrich",
+        "LookupJoinExpressionAfterLimitAndRemoteEnrich",
         // Lookup join after FORK is not support in CCS yet
         "ForkBeforeLookupJoin"
     );

.idea/runConfigurations/Debug_Elasticsearch__node_3_.xml

@@ -720,7 +720,11 @@ private LogicalPlan resolveLookup(Lookup l, List<Attribute> childrenOutput) {
             return l;
         }
 
-        private List<Expression> resolveJoinFilters(List<Expression> filters, List<Attribute> leftOutput, List<Attribute> rightOutput) {
+        private List<Expression> resolveJoinFiltersAndSwapIfNeeded(
+            List<Expression> filters,
+            List<Attribute> leftOutput,
+            List<Attribute> rightOutput
+        ) {
             if (filters.isEmpty()) {
                 return emptyList();
             }
@@ -729,11 +733,39 @@ private List<Expression> resolveJoinFilters(List<Expression> filters, List<Attri
 
             List<Expression> resolvedFilters = new ArrayList<>(filters.size());
             for (Expression filter : filters) {
-                resolvedFilters.add(filter.transformUp(UnresolvedAttribute.class, ua -> maybeResolveAttribute(ua, childrenOutput)));
+                Expression filterResolved = filter.transformUp(UnresolvedAttribute.class, ua -> maybeResolveAttribute(ua, childrenOutput));
+                resolvedFilters.add(resolveAndOrientJoinCondition(filterResolved, leftOutput, rightOutput));
             }
             return resolvedFilters;
         }
 
+        private Expression resolveAndOrientJoinCondition(Expression condition, List<Attribute> leftOutput, List<Attribute> rightOutput) {
+            if (condition instanceof EsqlBinaryComparison comp
+                && comp.left() instanceof Attribute leftAttr
+                && comp.right() instanceof Attribute rightAttr) {
+
+                boolean leftIsFromLeft = leftOutput.contains(leftAttr);
+                boolean rightIsFromRight = rightOutput.contains(rightAttr);
+
+                if (leftIsFromLeft && rightIsFromRight) {
+                    return comp; // Correct orientation
+                }
+
+                boolean leftIsFromRight = rightOutput.contains(leftAttr);
+                boolean rightIsFromLeft = leftOutput.contains(rightAttr);
+
+                if (leftIsFromRight && rightIsFromLeft) {
+                    return comp.swapLeftAndRight(); // Swapped orientation
+                }
+
+                // Invalid orientation (e.g., both from left or both from right)
+                throw new EsqlIllegalArgumentException(
+                    "Join condition must be between attributes on the left and right side, but found: " + condition.sourceText()
+                );
+            }
+            return condition; // Not a binary comparison between two attributes, no change needed.
+        }
+
         private Join resolveLookupJoin(LookupJoin join) {
             JoinConfig config = join.config();
             // for now, support only (LEFT) USING clauses
@@ -763,16 +795,19 @@ private Join resolveLookupJoin(LookupJoin join) {
                 List<Expression> resolvedFilters = new ArrayList<>();
                 List<Attribute> matchKeys;
                 if (join.config().joinOnConditions() != null) {
-                    resolvedFilters = resolveJoinFilters(
+                    resolvedFilters = resolveJoinFiltersAndSwapIfNeeded(
                         Predicates.splitAnd(join.config().joinOnConditions()),
                         join.left().output(),
                         join.right().output()
                     );
-                    // build leftKeys and rightKeys using the left side of the resolvedFilters.
+                    // build leftKeys and rightKeys using the correct side of the resolvedFilters.
+                    // resolveJoinFiltersAndSwapIfNeeded already put the left and right on the correct side
                     for (Expression expression : resolvedFilters) {
-                        if (expression instanceof EsqlBinaryComparison binaryComparison) {
-                            leftKeys.add((Attribute) binaryComparison.left());
-                            rightKeys.add((Attribute) binaryComparison.right());
+                        if (expression instanceof EsqlBinaryComparison binaryComparison
+                            && binaryComparison.left() instanceof Attribute leftAttribute
+                            && binaryComparison.right() instanceof Attribute rightAttribute) {
+                            leftKeys.add(leftAttribute);
+                            rightKeys.add(rightAttribute);
                         } else {
                             throw new EsqlIllegalArgumentException("Unsupported join filter expression: " + expression);
                         }

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

@@ -63,7 +63,7 @@ public ExpressionQueryList(
         if (queryLists.size() < 2 && (rightPreJoinPlan instanceof FilterExec == false) && request.getJoinOnConditions() == null) {
             throw new IllegalArgumentException("ExpressionQueryList must have at least two QueryLists or a pre-join filter");
         }
-        this.queryLists = queryLists;
+        this.queryLists = new ArrayList<>(queryLists);
         this.context = context;
         this.aliasFilter = aliasFilter;
         buildJoinOnConditions(request, clusterService, warnings);
@@ -80,10 +80,8 @@ private void buildJoinOnConditions(LookupFromIndexService.TransportRequest reque
             // the join on conditions are already populated via the queryLists
             // there is nothing to do here
             return;
-        } else {
-            // clear the join on conditions in the query lists
-            // the join on condition needs to come from the expression
-            queryLists.clear();
+        } else if (queryLists.isEmpty() == false) {
+            throw new IllegalArgumentException("ExpressionQueryList called with both join on expression and join on fields");
         }
         List<Expression> expressions = Predicates.splitAnd(filter);
         for (Expression expr : expressions) {

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

@@ -110,25 +110,28 @@ protected LookupEnrichQueryGenerator queryList(
         Block inputBlock,
         Warnings warnings
     ) {
-        List<QueryList> queryLists = new ArrayList<>();
-        for (int i = 0; i < request.matchFields.size(); i++) {
-            MatchConfig matchField = request.matchFields.get(i);
-            QueryList q = termQueryList(
-                context.getFieldType(matchField.fieldName().string()),
-                context,
-                aliasFilter,
-                request.inputPage.getBlock(matchField.channel()),
-                matchField.type()
-            ).onlySingleValues(warnings, "LOOKUP JOIN encountered multi-value");
-            queryLists.add(q);
-        }
+        if (request.joinOnConditions == null) {
+            List<QueryList> queryLists = new ArrayList<>();
+            for (int i = 0; i < request.matchFields.size(); i++) {
+                MatchConfig matchField = request.matchFields.get(i);
+                QueryList q = termQueryList(
+                    context.getFieldType(matchField.fieldName().string()),
+                    context,
+                    aliasFilter,
+                    request.inputPage.getBlock(matchField.channel()),
+                    matchField.type()
+                ).onlySingleValues(warnings, "LOOKUP JOIN encountered multi-value");
+                queryLists.add(q);
+            }
 
-        PhysicalPlan physicalPlan = request.rightPreJoinPlan;
-        physicalPlan = localLookupNodePlanning(physicalPlan);
-        if (queryLists.size() == 1 && physicalPlan instanceof FilterExec == false && request.joinOnConditions == null) {
-            return queryLists.getFirst();
+            PhysicalPlan physicalPlan = localLookupNodePlanning(request.rightPreJoinPlan);
+            if (queryLists.size() == 1 && physicalPlan instanceof FilterExec == false) {
+                return queryLists.getFirst();
+            }
+            return new ExpressionQueryList(queryLists, context, physicalPlan, clusterService, request, aliasFilter, warnings);
         }
-        return new ExpressionQueryList(queryLists, context, physicalPlan, clusterService, request, aliasFilter, warnings);
+        PhysicalPlan physicalPlan = localLookupNodePlanning(request.rightPreJoinPlan);
+        return new ExpressionQueryList(List.of(), context, physicalPlan, clusterService, request, aliasFilter, warnings);
 
     }
 

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

@@ -715,7 +715,6 @@ public JoinInfo visitExpressionBasedLookupJoin(EsqlBaseParser.ExpressionBasedLoo
                 );
             }
         }
-        validateJoinFields(joinFields);
         return new JoinInfo(joinFields, joinExpressions);
     }
 

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

@@ -232,23 +232,9 @@ public static List<Attribute> computeOutput(List<Attribute> leftOutput, List<Att
                     .toList();
                 output = mergeOutputAttributes(rightOutputWithoutMatchFields, leftOutput);
             } else {
-                // suppose we have
-                // index1 a, b, c, d
-                // index2 c, d, e, f
-                // we have expression join on (index1.b > index2.c AND index1.d > index2.d)
-                // We keep a,b,d from the left side (a, b only on left, d is join key, so we keep it from the left)
-                // we keep c,e,f from the right side(c is not part of the join key on the left so gets shadowed, e,f only on right)
-                Set<String> leftJoinKeyNames = config.leftFields()
-                    .stream()
-                    .map(Attribute::name)
-                    .collect(java.util.stream.Collectors.toSet());
-                // as we can do (left_key > right_key) now as join condition, we want to preserve the right_key
-                // unless it is also a key on the left side,
-                // but we need to do name equality only
-                List<Attribute> rightOutputWithoutMatchFields = rightOutput.stream()
-                    .filter(attr -> leftJoinKeyNames.contains(attr.name()) == false)
-                    .toList();
-                output = mergeOutputAttributes(rightOutputWithoutMatchFields, leftOutput);
+                // We don't allow any attributes in the joinOnConditions that don't have unique names
+                // so right always overwrites left in case of name clashes
+                output = mergeOutputAttributes(rightOutput, leftOutput);
             }
 
         } else {

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

@@ -8994,4 +8994,65 @@ public void testTranslateMetricsGroupedByTBucketInTSMode() {
         Bucket bucket = as(Alias.unwrap(eval.fields().get(0)), Bucket.class);
         assertThat(Expressions.attribute(bucket.field()).name(), equalTo("@timestamp"));
     }
+
+    /**
+     * Limit[1000[INTEGER],true]
+     * \_Join[LEFT,[languages{f}#8, language_code{f}#16],[languages{f}#8],[language_code{f}#16],languages{f}#8 == language_code{
+     * f}#16]
+     *   |_EsqlProject[[languages{f}#8]]
+     *   | \_Limit[1000[INTEGER],false]
+     *   |   \_EsRelation[test][_meta_field{f}#11, emp_no{f}#5, first_name{f}#6, ge..]
+     *   \_EsRelation[languages_lookup][LOOKUP][language_code{f}#16, language_name{f}#17]
+     */
+    public void testLookupJoinExpressionSwapped() {
+        LogicalPlan plan = optimizedPlan("""
+            from test
+            | keep languages
+            | lookup join languages_lookup ON language_code == languages
+            """);
+
+        var limit = asLimit(plan, 1000, true);
+        var join = as(limit.child(), Join.class);
+        assertEquals("language_code == languages", join.config().joinOnConditions().toString());
+        var equals = as(join.config().joinOnConditions(), Equals.class);
+        // we expect left and right to be swapped
+        var left = as(equals.left(), Attribute.class);
+        var right = as(equals.right(), Attribute.class);
+        assertEquals("language_code", right.name());
+        assertEquals("languages", left.name());
+        var project = as(join.left(), EsqlProject.class);
+        var limitPastJoin = asLimit(project.child(), 1000, false);
+        as(limitPastJoin.child(), EsRelation.class);
+        as(join.right(), EsRelation.class);
+    }
+
+    public void testLookupJoinExpressionAmbigiousRight() {
+        String query = """
+            from test
+            | rename languages as language_code
+            | lookup join languages_lookup ON salary == language_code
+            """;
+        IllegalStateException e = expectThrows(IllegalStateException.class, () -> plan(query));
+        assertThat(e.getMessage(), containsString("Reference [language_code] is ambiguous; matches any of "));
+    }
+
+    public void testLookupJoinExpressionAmbigiousLeft() {
+        String query = """
+            from test
+            | rename languages as language_name
+            | lookup join languages_lookup ON language_name == language_code
+            """;
+        IllegalStateException e = expectThrows(IllegalStateException.class, () -> plan(query));
+        assertThat(e.getMessage(), containsString("Reference [language_name] is ambiguous; matches any of "));
+    }
+
+    public void testLookupJoinExpressionAmbigiousBoth() {
+        String query = """
+            from test
+            | rename languages as language_code
+            | lookup join languages_lookup ON language_code != language_code
+            """;
+        IllegalStateException e = expectThrows(IllegalStateException.class, () -> plan(query));
+        assertThat(e.getMessage(), containsString("Reference [language_code] is ambiguous; matches any of "));
+    }
 }