ESQL: Fix LOOKUP attribute shadowing (#109807)

Fix https://github.com/elastic/elasticsearch/issues/109392

This makes attribute shadowing of LOOKUP consistent with ENRICH,
DISSECT/GROK and EVAL.

docs/changelog/109807.yaml

@@ -0,0 +1,6 @@
+pr: 109807
+summary: "ESQL: Fix LOOKUP attribute shadowing"
+area: ES|QL
+type: bug
+issues:
+ - 109392

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

@@ -14,6 +14,43 @@ foo bar   | null       | null
 ;
 
 
+shadowing
+FROM employees
+| KEEP first_name, last_name
+| WHERE last_name == "Facello"
+| EVAL left = "left", full_name = concat(first_name, " ", last_name) , last_name = "last_name", right = "right"
+| DISSECT full_name "%{?} %{last_name}"
+;
+
+first_name:keyword | left:keyword | full_name:keyword | right:keyword | last_name:keyword
+Georgi             | left         |    Georgi Facello | right         | Facello
+;
+
+shadowingSelf
+FROM employees
+| KEEP first_name, last_name
+| WHERE last_name == "Facello"
+| EVAL left = "left", name = concat(first_name, "1 ", last_name), right = "right"
+| DISSECT name "%{name} %{?}"
+;
+
+first_name:keyword | last_name:keyword | left:keyword | right:keyword | name:keyword
+Georgi             | Facello           | left         | right         | Georgi1
+;
+
+shadowingMulti
+FROM employees
+| KEEP first_name, last_name
+| WHERE last_name == "Facello"
+| EVAL left = "left", foo = concat(first_name, "1 ", first_name, "2 ", last_name) , middle = "middle", bar = "bar", right = "right"
+| DISSECT foo "%{bar} %{first_name} %{last_name_again}"
+;
+
+last_name:keyword | left:keyword | foo:keyword             | middle:keyword | right:keyword | bar:keyword  | first_name:keyword | last_name_again:keyword
+Facello           | left         | Georgi1 Georgi2 Facello | middle         | right         | Georgi1      | Georgi2            | Facello
+;
+
+
 complexPattern
 ROW a = "1953-01-23T12:15:00Z - some text - 127.0.0.1;" 
 | DISSECT a "%{Y}-%{M}-%{D}T%{h}:%{m}:%{s}Z - %{msg} - %{ip};" 

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

@@ -31,6 +31,82 @@ FROM sample_data
 median_duration:double | env:keyword
 ;
 
+shadowing
+required_capability: enrich_load
+ROW left = "left", client_ip = "172.21.0.5", env = "env", right = "right"
+| ENRICH clientip_policy ON client_ip
+;
+
+left:keyword | client_ip:keyword | right:keyword | env:keyword
+left         | 172.21.0.5        | right         | Development
+;
+
+shadowingLimit0
+ROW left = "left", client_ip = "172.21.0.5", env = "env", right = "right"
+| ENRICH clientip_policy ON client_ip
+| LIMIT 0
+;
+
+left:keyword | client_ip:keyword | right:keyword | env:keyword
+;
+
+shadowingWithAlias
+required_capability: enrich_load
+ROW left = "left", foo = "foo", client_ip = "172.21.0.5", env = "env", right = "right"
+| ENRICH clientip_policy ON client_ip WITH foo = env
+;
+
+left:keyword | client_ip:keyword | env:keyword | right:keyword | foo:keyword
+left         | 172.21.0.5        | env         | right         | Development
+;
+
+shadowingWithAliasLimit0
+ROW left = "left", foo = "foo", client_ip = "172.21.0.5", env = "env", right = "right"
+| ENRICH clientip_policy ON client_ip WITH foo = env
+| LIMIT 0
+;
+
+left:keyword | client_ip:keyword | env:keyword | right:keyword | foo:keyword
+;
+
+shadowingSelf
+required_capability: enrich_load
+ROW left = "left", client_ip = "172.21.0.5", env = "env", right = "right"
+| ENRICH clientip_policy ON client_ip WITH client_ip = env
+;
+
+left:keyword | env:keyword | right:keyword | client_ip:keyword
+left         | env         | right         | Development
+;
+
+shadowingSelfLimit0
+ROW left = "left", client_ip = "172.21.0.5", env = "env", right = "right"
+| ENRICH clientip_policy ON client_ip WITH client_ip = env
+| LIMIT 0
+;
+
+left:keyword | env:keyword | right:keyword | client_ip:keyword
+;
+
+shadowingMulti
+required_capability: enrich_load
+ROW left = "left", airport = "Zurich Airport ZRH", city = "Zürich", middle = "middle", region = "North-East Switzerland", right = "right" 
+| ENRICH city_names ON city WITH airport, region, city_boundary
+;
+
+left:keyword | city:keyword | middle:keyword | right:keyword | airport:text | region:text   | city_boundary:geo_shape
+left         | Zürich       | middle         | right         | Zurich Int'l | Bezirk Zürich | "POLYGON((8.448 47.3802,8.4977 47.3452,8.5032 47.3202,8.6254 47.3547,8.5832 47.3883,8.5973 47.4063,8.5431 47.4329,8.4858 47.431,8.4691 47.4169,8.473 47.3951,8.448 47.3802))"
+;
+
+shadowingMultiLimit0
+ROW left = "left", airport = "Zurich Airport ZRH", city = "Zurich", middle = "middle", region = "North-East Switzerland", right = "right" 
+| ENRICH city_names ON city WITH airport, region, city_boundary
+| LIMIT 0
+;
+
+left:keyword | city:keyword | middle:keyword | right:keyword | airport:text | region:text | city_boundary:geo_shape
+;
+
 simple
 required_capability: enrich_load
 

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

@@ -5,6 +5,34 @@ a:integer | b:integer
 1         | 2
 ;
 
+
+shadowing
+ROW left = "left", x = 10000 , right = "right"
+| EVAL x = 1
+;
+
+left:keyword | right:keyword | x:integer
+left         | right         | 1
+;
+
+shadowingSelf
+ROW left = "left", x = 10000 , right = "right"
+| EVAL x = x + 1
+;
+
+left:keyword | right:keyword | x:integer
+left         | right         | 10001
+;
+
+shadowingMulti
+ROW left = "left", x = 0, middle = "middle", y = -1, right = "right"
+| EVAL x = 9, y = 10
+;
+
+left:keyword | middle:keyword | right:keyword | x:integer | y:integer
+left         | middle         | right         | 9         | 10
+;
+
 withMath
 row a = 1 | eval b = 2 + 3;
 

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

@@ -14,6 +14,43 @@ foo bar   | null
 ;
 
 
+shadowing
+FROM employees
+| KEEP first_name, last_name
+| WHERE last_name == "Facello"
+| EVAL left = "left", full_name = concat(first_name, " ", last_name) , last_name = "last_name", right = "right"
+| GROK full_name "%{WORD} %{WORD:last_name}"
+;
+
+first_name:keyword | left:keyword | full_name:keyword | right:keyword | last_name:keyword
+Georgi             | left         |    Georgi Facello | right         | Facello
+;
+
+shadowingSelf
+FROM employees
+| KEEP first_name, last_name
+| WHERE last_name == "Facello"
+| EVAL left = "left", name = concat(first_name, "1 ", last_name), right = "right"
+| GROK name "%{WORD:name} %{WORD}"
+;
+
+first_name:keyword | last_name:keyword | left:keyword | right:keyword | name:keyword
+Georgi             | Facello           | left         | right         | Georgi1
+;
+
+shadowingMulti
+FROM employees
+| KEEP first_name, last_name
+| WHERE last_name == "Facello"
+| EVAL left = "left", foo = concat(first_name, "1 ", first_name, "2 ", last_name) , middle = "middle", bar = "bar", right = "right"
+| GROK foo "%{WORD:bar} %{WORD:first_name} %{WORD:last_name_again}"
+;
+
+last_name:keyword | left:keyword | foo:keyword             | middle:keyword | right:keyword | bar:keyword  | first_name:keyword | last_name_again:keyword
+Facello           | left         | Georgi1 Georgi2 Facello | middle         | right         | Georgi1      | Georgi2            | Facello
+;
+
+
 complexPattern
 ROW a = "1953-01-23T12:15:00Z 127.0.0.1 some.email@foo.com 42" 
 | GROK a "%{TIMESTAMP_ISO8601:date} %{IP:ip} %{EMAILADDRESS:email} %{NUMBER:num:int}" 

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

@@ -1,5 +1,5 @@
 keywordByInt
-required_capability: tables_types
+required_capability: lookup_v3
 FROM employees
 | SORT emp_no
 | LIMIT 4
@@ -17,7 +17,7 @@ emp_no:integer | languages:integer | lang_name:keyword
 ;
 
 keywordByMvInt
-required_capability: tables_types
+required_capability: lookup_v3
 ROW int=[1, 2, 3]
 | LOOKUP int_number_names ON int
 ;
@@ -27,7 +27,7 @@ int:integer | name:keyword
 ;
 
 keywordByDupeInt
-required_capability: tables_types
+required_capability: lookup_v3
 ROW int=[1, 1, 1]
 | LOOKUP int_number_names ON int
 ;
@@ -37,7 +37,7 @@ int:integer | name:keyword
 ;
 
 intByKeyword
-required_capability: tables_types
+required_capability: lookup_v3
 ROW name="two"
 | LOOKUP int_number_names ON name
 ;
@@ -48,7 +48,7 @@ name:keyword | int:integer
 
 
 keywordByLong
-required_capability: tables_types
+required_capability: lookup_v3
 FROM employees
 | SORT emp_no
 | LIMIT 4
@@ -66,7 +66,7 @@ emp_no:integer | languages:long | lang_name:keyword
 ;
 
 longByKeyword
-required_capability: tables_types
+required_capability: lookup_v3
 ROW name="two"
 | LOOKUP long_number_names ON name
 ;
@@ -76,7 +76,7 @@ name:keyword | long:long
 ;
 
 keywordByFloat
-required_capability: tables_types
+required_capability: lookup_v3
 FROM employees
 | SORT emp_no
 | LIMIT 4
@@ -94,7 +94,7 @@ emp_no:integer | height:double | height_name:keyword
 ;
 
 floatByKeyword
-required_capability: tables_types
+required_capability: lookup_v3
 ROW name="two point zero eight"
 | LOOKUP double_number_names ON name
 ;
@@ -104,7 +104,7 @@ two point zero eight |          2.08
 ;
 
 floatByNullMissing
-required_capability: tables_types
+required_capability: lookup_v3
 ROW name=null
 | LOOKUP double_number_names ON name
 ;
@@ -114,7 +114,7 @@ name:null | double:double
 ;
 
 floatByNullMatching
-required_capability: tables_types
+required_capability: lookup_v3
 ROW name=null
 | LOOKUP double_number_names_with_null ON name
 ;
@@ -124,7 +124,7 @@ name:null | double:double
 ;
 
 intIntByKeywordKeyword
-required_capability: tables_types
+required_capability: lookup_v3
 ROW aa="foo", ab="zoo"
 | LOOKUP big ON aa, ab
 ;
@@ -134,7 +134,7 @@ foo        | zoo        |          1 |         -1
 ;
 
 intIntByKeywordKeywordMissing
-required_capability: tables_types
+required_capability: lookup_v3
 ROW aa="foo", ab="zoi"
 | LOOKUP big ON aa, ab
 ;
@@ -144,7 +144,7 @@ foo        | zoi        |       null |       null
 ;
 
 intIntByKeywordKeywordSameValues
-required_capability: tables_types
+required_capability: lookup_v3
 ROW aa="foo", ab="foo"
 | LOOKUP big ON aa, ab
 ;
@@ -154,7 +154,7 @@ foo        | foo        |          2 |         -2
 ;
 
 intIntByKeywordKeywordSameValuesMissing
-required_capability: tables_types
+required_capability: lookup_v3
 ROW aa="bar", ab="bar"
 | LOOKUP big ON aa, ab
 ;
@@ -164,7 +164,7 @@ bar        | bar        |       null |       null
 ;
 
 lookupBeforeStats
-required_capability: tables_types
+required_capability: lookup_v3
   FROM employees
 | RENAME languages AS int
 | LOOKUP int_number_names ON int
@@ -182,7 +182,7 @@ height:double | languages:keyword
 ;
 
 lookupAfterStats
-required_capability: tables_types
+required_capability: lookup_v3
   FROM employees
 | STATS int=TO_INT(AVG(height))
 | LOOKUP int_number_names ON int
@@ -194,7 +194,7 @@ two
 
 // Makes sure the LOOKUP squashes previous names 
 doesNotDuplicateNames
-required_capability: tables_types
+required_capability: lookup_v3
 FROM employees
 | SORT emp_no
 | LIMIT 4
@@ -213,7 +213,7 @@ emp_no:integer | languages:long | name:keyword
 ;
 
 lookupBeforeSort
-required_capability: tables_types
+required_capability: lookup_v3
 FROM employees
 | WHERE emp_no < 10005
 | RENAME languages AS int
@@ -231,7 +231,7 @@ languages:keyword | emp_no:integer
 ;
 
 lookupAfterSort
-required_capability: tables_types
+required_capability: lookup_v3
 FROM employees
 | WHERE emp_no < 10005
 | SORT languages ASC, emp_no ASC
@@ -248,12 +248,38 @@ languages:keyword | emp_no:integer
              five | 10004
 ;
 
+shadowing
+required_capability: lookup_v3
+FROM employees
+| KEEP emp_no
+| WHERE emp_no == 10001
+| EVAL left = "left", int = emp_no - 10000, name = "name", right = "right"
+| LOOKUP int_number_names ON int
+;
+
+emp_no:integer | left:keyword | int:integer | right:keyword | name:keyword
+         10001 | left         |           1 | right         | one
+;
+
+shadowingMulti
+required_capability: lookup_v3
+FROM employees
+| KEEP emp_no
+| WHERE emp_no == 10001
+| EVAL left = "left", nb = -10011+emp_no, na = "na", middle = "middle", ab = "ab", aa = "bar", right = "right"
+| LOOKUP big ON aa, nb
+;
+
+emp_no:integer | left:keyword | nb:integer | middle:keyword | aa:keyword | right:keyword | ab:keyword | na:integer 
+         10001 | left         |        -10 | middle         | bar        | right         | zop        | 10
+;
+
 //
 // Make sure that the new LOOKUP syntax doesn't clash with any existing things
 // named "lookup"
 //
 rowNamedLookup
-required_capability: tables_types
+required_capability: lookup_v3
 ROW lookup = "a"
 ;
 
@@ -262,7 +288,7 @@ lookup:keyword
 ;
 
 rowNamedLOOKUP
-required_capability: tables_types
+required_capability: lookup_v3
 ROW LOOKUP = "a"
 ;
 
@@ -271,7 +297,7 @@ LOOKUP:keyword
 ;
 
 evalNamedLookup
-required_capability: tables_types
+required_capability: lookup_v3
 ROW a = "a" | EVAL lookup = CONCAT(a, "1")
 ;
 
@@ -280,7 +306,7 @@ a:keyword | lookup:keyword
 ;
 
 dissectNamedLookup
-required_capability: tables_types
+required_capability: lookup_v3
 row a = "foo bar" | dissect a "foo %{lookup}";
 
 a:keyword | lookup:keyword
@@ -288,7 +314,7 @@ a:keyword | lookup:keyword
 ;
 
 renameIntoLookup
-required_capability: tables_types
+required_capability: lookup_v3
 row a = "foo bar" | RENAME a AS lookup;
 
 lookup:keyword
@@ -296,7 +322,7 @@ lookup:keyword
 ;
 
 sortOnLookup
-required_capability: tables_types
+required_capability: lookup_v3
 ROW lookup = "a" | SORT lookup
 ;
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -60,9 +60,11 @@ public class EsqlCapabilities {
         METADATA_IGNORED_FIELD,
 
         /**
-         * Support for the syntax {@code "tables": {"type": [<values>]}}.
+         * LOOKUP command with
+         * - tables using syntax {@code "tables": {"type": [<values>]}}
+         * - fixed variable shadowing
          */
-        TABLES_TYPES(true),
+        LOOKUP_V3(true),
 
         /**
          * Support for requesting the "REPEAT" command.

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

@@ -519,13 +519,13 @@ public class Analyzer extends ParameterizedRuleExecutor<LogicalPlan, AnalyzerCon
             }
 
             // check the on field against both the child output and the inner relation
-            List<NamedExpression> matchFields = new ArrayList<>(l.matchFields().size());
+            List<Attribute> matchFields = new ArrayList<>(l.matchFields().size());
             List<Attribute> localOutput = l.localRelation().output();
             boolean modified = false;
 
-            for (NamedExpression ne : l.matchFields()) {
-                NamedExpression matchFieldChildReference = ne;
-                if (ne instanceof UnresolvedAttribute ua && ua.customMessage() == false) {
+            for (Attribute matchField : l.matchFields()) {
+                Attribute matchFieldChildReference = matchField;
+                if (matchField instanceof UnresolvedAttribute ua && ua.customMessage() == false) {
                     modified = true;
                     Attribute joinedAttribute = maybeResolveAttribute(ua, localOutput);
                     // can't find the field inside the local relation

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/LogicalPlanBuilder.java

@@ -453,11 +453,17 @@ public class LogicalPlanBuilder extends ExpressionBuilder {
         }
         var source = source(ctx);
 
-        List<NamedExpression> matchFields = visitQualifiedNamePatterns(ctx.qualifiedNamePatterns(), ne -> {
+        @SuppressWarnings("unchecked")
+        List<Attribute> matchFields = (List<Attribute>) (List) visitQualifiedNamePatterns(ctx.qualifiedNamePatterns(), ne -> {
             if (ne instanceof UnresolvedNamePattern || ne instanceof UnresolvedStar) {
                 var src = ne.source();
                 throw new ParsingException(src, "Using wildcards [*] in LOOKUP ON is not allowed yet [{}]", src.text());
             }
+            if ((ne instanceof UnresolvedAttribute) == false) {
+                throw new IllegalStateException(
+                    "visitQualifiedNamePatterns can only return UnresolvedNamePattern, UnresolvedStar or UnresolvedAttribute"
+                );
+            }
         });
 
         Literal tableName = new Literal(source, ctx.tableName.getText(), DataType.KEYWORD);

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

@@ -11,7 +11,6 @@ import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xpack.esql.core.capabilities.Resolvables;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
-import org.elasticsearch.xpack.esql.core.expression.Expressions;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.core.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.esql.core.plan.logical.UnaryPlan;
@@ -39,7 +38,7 @@ public class Lookup extends UnaryPlan {
     /**
      * References to the input fields to match against the {@link #localRelation}.
      */
-    private final List<NamedExpression> matchFields;
+    private final List<Attribute> matchFields;
     // initialized during the analysis phase for output and validation
     // afterward, it is converted into a Join (BinaryPlan) hence why here it is not a child
     private final LocalRelation localRelation;
@@ -49,7 +48,7 @@ public class Lookup extends UnaryPlan {
         Source source,
         LogicalPlan child,
         Expression tableName,
-        List<NamedExpression> matchFields,
+        List<Attribute> matchFields,
         @Nullable LocalRelation localRelation
     ) {
         super(source, child);
@@ -61,7 +60,7 @@ public class Lookup extends UnaryPlan {
     public Lookup(PlanStreamInput in) throws IOException {
         super(Source.readFrom(in), in.readLogicalPlanNode());
         this.tableName = in.readExpression();
-        this.matchFields = in.readNamedWriteableCollectionAsList(NamedExpression.class);
+        this.matchFields = in.readNamedWriteableCollectionAsList(Attribute.class);
         this.localRelation = in.readBoolean() ? new LocalRelation(in) : null;
     }
 
@@ -82,7 +81,7 @@ public class Lookup extends UnaryPlan {
         return tableName;
     }
 
-    public List<NamedExpression> matchFields() {
+    public List<Attribute> matchFields() {
         return matchFields;
     }
 
@@ -122,10 +121,10 @@ public class Lookup extends UnaryPlan {
     @Override
     public List<Attribute> output() {
         if (lazyOutput == null) {
-            List<Attribute> rightSide = localRelation != null
-                ? Join.makeNullable(Join.makeReference(localRelation.output()))
-                : Expressions.asAttributes(matchFields);
-            lazyOutput = Join.mergeOutput(child().output(), rightSide, matchFields);
+            if (localRelation == null) {
+                throw new IllegalStateException("Cannot determine output of LOOKUP with unresolved table");
+            }
+            lazyOutput = Join.computeOutput(child().output(), localRelation.output(), joinConfig());
         }
         return lazyOutput;
     }

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

@@ -8,7 +8,8 @@
 package org.elasticsearch.xpack.esql.plan.logical.join;
 
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
-import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
+import org.elasticsearch.xpack.esql.core.expression.AttributeSet;
+import org.elasticsearch.xpack.esql.core.expression.Expressions;
 import org.elasticsearch.xpack.esql.core.expression.Nullability;
 import org.elasticsearch.xpack.esql.core.expression.ReferenceAttribute;
 import org.elasticsearch.xpack.esql.core.plan.logical.BinaryPlan;
@@ -20,8 +21,12 @@ import org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
+import java.util.Set;
+
+import static org.elasticsearch.xpack.esql.expression.NamedExpressions.mergeOutputAttributes;
 
 public class Join extends BinaryPlan {
 
@@ -68,47 +73,41 @@ public class Join extends BinaryPlan {
     @Override
     public List<Attribute> output() {
         if (lazyOutput == null) {
-            lazyOutput = computeOutput();
+            lazyOutput = computeOutput(left().output(), right().output(), config);
         }
         return lazyOutput;
     }
 
-    private List<Attribute> computeOutput() {
-        List<Attribute> right = makeReference(right().output());
+    /**
+     * Merge output fields.
+     * Currently only implemented for LEFT JOINs; the rightOutput shadows the leftOutput, except for any attributes that
+     * occur in the join's matchFields.
+     */
+    public static List<Attribute> computeOutput(List<Attribute> leftOutput, List<Attribute> rightOutput, JoinConfig config) {
+        AttributeSet matchFieldSet = new AttributeSet(config.matchFields());
+        Set<String> matchFieldNames = new HashSet<>(Expressions.names(config.matchFields()));
         return switch (config.type()) {
-            case LEFT -> // right side becomes nullable
-                mergeOutput(left().output(), makeNullable(right), config.matchFields());
-            case RIGHT -> // left side becomes nullable
-                mergeOutput(makeNullable(left().output()), right, config.matchFields());
-            case FULL -> // both sides become nullable
-                mergeOutput(makeNullable(left().output()), makeNullable(right), config.matchFields());
-            default -> // neither side becomes nullable
-                mergeOutput(left().output(), right, config.matchFields());
+            case LEFT -> {
+                // Right side becomes nullable.
+                List<Attribute> fieldsAddedFromRight = removeCollisionsWithMatchFields(rightOutput, matchFieldSet, matchFieldNames);
+                yield mergeOutputAttributes(makeNullable(makeReference(fieldsAddedFromRight)), leftOutput);
+            }
+            default -> throw new UnsupportedOperationException("Other JOINs than LEFT not supported");
         };
     }
 
-    /**
-     * Merge output fields, left hand side wins in name conflicts <strong>except</strong>
-     * for fields defined in {@link JoinConfig#matchFields()}.
-     */
-    public static List<Attribute> mergeOutput(
-        List<? extends Attribute> lhs,
-        List<? extends Attribute> rhs,
-        List<NamedExpression> matchFields
+    private static List<Attribute> removeCollisionsWithMatchFields(
+        List<Attribute> attributes,
+        AttributeSet matchFields,
+        Set<String> matchFieldNames
     ) {
-        List<Attribute> results = new ArrayList<>(lhs.size() + rhs.size());
-
-        for (Attribute a : lhs) {
-            if (rhs.contains(a) == false || matchFields.stream().anyMatch(m -> m.name().equals(a.name()))) {
-                results.add(a);
-            }
-        }
-        for (Attribute a : rhs) {
-            if (false == matchFields.stream().anyMatch(m -> m.name().equals(a.name()))) {
-                results.add(a);
+        List<Attribute> result = new ArrayList<>();
+        for (Attribute attr : attributes) {
+            if ((matchFields.contains(attr) || matchFieldNames.contains(attr.name())) == false) {
+                result.add(attr);
             }
         }
-        return results;
+        return result;
     }
 
     /**
@@ -125,7 +124,7 @@ public class Join extends BinaryPlan {
     public static List<Attribute> makeReference(List<Attribute> output) {
         List<Attribute> out = new ArrayList<>(output.size());
         for (Attribute a : output) {
-            if (a.resolved()) {
+            if (a.resolved() && a instanceof ReferenceAttribute == false) {
                 out.add(new ReferenceAttribute(a.source(), a.name(), a.dataType(), a.qualifier(), a.nullable(), a.id(), a.synthetic()));
             } else {
                 out.add(a);

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

@@ -11,8 +11,8 @@ import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.xpack.esql.core.capabilities.Resolvables;
+import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
-import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput;
 
@@ -24,11 +24,11 @@ import java.util.List;
  * @param matchFields fields that are merged from the left and right relations
  * @param conditions when these conditions are true the rows are joined
  */
-public record JoinConfig(JoinType type, List<NamedExpression> matchFields, List<Expression> conditions) implements Writeable {
+public record JoinConfig(JoinType type, List<Attribute> matchFields, List<Expression> conditions) implements Writeable {
     public JoinConfig(StreamInput in) throws IOException {
         this(
             JoinType.readFrom(in),
-            in.readNamedWriteableCollectionAsList(NamedExpression.class),
+            in.readNamedWriteableCollectionAsList(Attribute.class),
             in.readCollectionAsList(i -> ((PlanStreamInput) i).readExpression())
         );
     }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/HashJoinExec.java

@@ -9,7 +9,6 @@ package org.elasticsearch.xpack.esql.plan.physical;
 
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.AttributeSet;
-import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.Equals;
@@ -24,7 +23,7 @@ import java.util.Set;
 
 public class HashJoinExec extends UnaryExec implements EstimatesRowSize {
     private final LocalSourceExec joinData;
-    private final List<NamedExpression> matchFields;
+    private final List<Attribute> matchFields;
     /**
      * Conditions that must match for rows to be joined. The {@link Equals#left()}
      * is always from the child and the {@link Equals#right()} is always from the
@@ -38,7 +37,7 @@ public class HashJoinExec extends UnaryExec implements EstimatesRowSize {
         Source source,
         PhysicalPlan child,
         LocalSourceExec hashData,
-        List<NamedExpression> matchFields,
+        List<Attribute> matchFields,
         List<Equals> conditions,
         List<Attribute> output
     ) {
@@ -52,7 +51,7 @@ public class HashJoinExec extends UnaryExec implements EstimatesRowSize {
     public HashJoinExec(PlanStreamInput in) throws IOException {
         super(Source.readFrom(in), in.readPhysicalPlanNode());
         this.joinData = new LocalSourceExec(in);
-        this.matchFields = in.readNamedWriteableCollectionAsList(NamedExpression.class);
+        this.matchFields = in.readNamedWriteableCollectionAsList(Attribute.class);
         this.conditions = in.readCollectionAsList(i -> (Equals) EsqlBinaryComparison.readFrom(in));
         this.output = in.readNamedWriteableCollectionAsList(Attribute.class);
     }
@@ -70,7 +69,7 @@ public class HashJoinExec extends UnaryExec implements EstimatesRowSize {
         return joinData;
     }
 
-    public List<NamedExpression> matchFields() {
+    public List<Attribute> matchFields() {
         return matchFields;
     }
 

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/tree/EsqlNodeSubclassTests.java

@@ -12,10 +12,10 @@ import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.dissect.DissectParser;
 import org.elasticsearch.xpack.esql.core.capabilities.UnresolvedException;
+import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
-import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.core.expression.Order;
 import org.elasticsearch.xpack.esql.core.expression.UnresolvedAttribute;
 import org.elasticsearch.xpack.esql.core.expression.UnresolvedNamedExpression;
@@ -89,7 +89,7 @@ public class EsqlNodeSubclassTests<T extends B, B extends Node<B>> extends NodeS
         } else if (argClass == JoinConfig.class) {
             return new JoinConfig(
                 randomFrom(JoinType.values()),
-                randomList(0, 10, () -> (NamedExpression) makeArg(NamedExpression.class)),
+                randomList(0, 10, () -> (Attribute) makeArg(Attribute.class)),
                 randomList(0, 10, () -> (Expression) makeArg(Expression.class))
             );
         }