ESQL: Fix variable shadowing when pushing down past Project

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

@@ -575,6 +575,43 @@ ca:l | cx:l | l:i
 1    | 1    | null
 ;
 
+countSameFieldWithEval
+from employees | stats  b = count(gender), c = count(gender) by gender | eval b = gender
+;
+
+c:l | gender:s | b:s
+0   | null     | null
+57  | M        | M
+33  | F        | F
+;
+
+countSameFieldWithDissect
+from employees | stats b = count(gender), c = count(gender) by gender | dissect gender "%{b}"
+;
+
+c:l | gender:s | b:s
+0   | null     | null
+57  | M        | M
+33  | F        | F
+;
+
+countSameFieldWithGrok
+from employees | stats  b = count(gender), c = count(gender) by gender | grok gender "%{USERNAME:b}"
+;
+
+c:l | gender:s | b:s
+0   | null     | null
+57  | M        | M
+33  | F        | F
+;
+
+countSameFieldWithEnrich
+from employees | stats  b = count(gender), c = count(gender) by gender | enrich languages_policy on gender with b = language_name | limit 0
+;
+
+c:l | gender:s | b:s
+;
+
 aggsWithoutStats
 from employees | stats by gender | sort gender;
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/NamedExpressions.java

@@ -34,7 +34,8 @@ public static List<Attribute> mergeOutputAttributes(
     /**
      * Merges output expressions of a command given the new attributes plus the existing inputs that are emitted as outputs.
      * As a general rule, child output will come first in the list, followed by the new fields.
-     * In case of name collisions, only last entry is preserved (previous expressions with the same name are discarded)
+     * In case of name collisions, only the last entry is preserved (previous expressions with the same name are discarded)
+     * and the new attributes have precedence over the child output.
      * @param fields the fields added by the command
      * @param childOutput the command input that has to be propagated as output
      * @return

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/io/stream/PlanNamedTypes.java

@@ -474,15 +474,17 @@ static void writeAggregateExec(PlanStreamOutput out, AggregateExec aggregateExec
     }
 
     static DissectExec readDissectExec(PlanStreamInput in) throws IOException {
-        return new DissectExec(in.readSource(), in.readPhysicalPlanNode(), in.readExpression(), readDissectParser(in), readAttributes(in));
+        // TODO: needs new transport version
+        return new DissectExec(in.readSource(), in.readPhysicalPlanNode(), in.readExpression(), readDissectParser(in), readAliases(in));
     }
 
     static void writeDissectExec(PlanStreamOutput out, DissectExec dissectExec) throws IOException {
         out.writeNoSource();
         out.writePhysicalPlanNode(dissectExec.child());
         out.writeExpression(dissectExec.inputExpression());
         writeDissectParser(out, dissectExec.parser());
-        writeAttributes(out, dissectExec.extractedFields());
+        // TODO: needs new transport version
+        writeAliases(out, dissectExec.extractedFields());
     }
 
     static EsQueryExec readEsQueryExec(PlanStreamInput in) throws IOException {
@@ -665,7 +667,8 @@ static GrokExec readGrokExec(PlanStreamInput in) throws IOException {
             in.readPhysicalPlanNode(),
             in.readExpression(),
             Grok.pattern(source, in.readString()),
-            readAttributes(in)
+            // TODO: needs new transport version
+            readAliases(in)
         );
     }
 
@@ -674,7 +677,8 @@ static void writeGrokExec(PlanStreamOutput out, GrokExec grokExec) throws IOExce
         out.writePhysicalPlanNode(grokExec.child());
         out.writeExpression(grokExec.inputExpression());
         out.writeString(grokExec.pattern().pattern());
-        writeAttributes(out, grokExec.extractedFields());
+        // TODO: needs new transport version
+        writeAliases(out, grokExec.extractedFields());
     }
 
     static LimitExec readLimitExec(PlanStreamInput in) throws IOException {
@@ -779,15 +783,17 @@ static void writeAggregate(PlanStreamOutput out, Aggregate aggregate) throws IOE
     }
 
     static Dissect readDissect(PlanStreamInput in) throws IOException {
-        return new Dissect(in.readSource(), in.readLogicalPlanNode(), in.readExpression(), readDissectParser(in), readAttributes(in));
+        // TODO: needs new transport version
+        return new Dissect(in.readSource(), in.readLogicalPlanNode(), in.readExpression(), readDissectParser(in), readAliases(in));
     }
 
     static void writeDissect(PlanStreamOutput out, Dissect dissect) throws IOException {
         out.writeNoSource();
         out.writeLogicalPlanNode(dissect.child());
         out.writeExpression(dissect.input());
         writeDissectParser(out, dissect.parser());
-        writeAttributes(out, dissect.extractedFields());
+        // TODO: needs new transport version
+        writeAliases(out, dissect.extractedFields());
     }
 
     static EsRelation readEsRelation(PlanStreamInput in) throws IOException {
@@ -903,7 +909,8 @@ static Grok readGrok(PlanStreamInput in) throws IOException {
             in.readLogicalPlanNode(),
             in.readExpression(),
             Grok.pattern(source, in.readString()),
-            readAttributes(in)
+            // TODO: needs new transport version
+            readAliases(in)
         );
     }
 
@@ -912,7 +919,8 @@ static void writeGrok(PlanStreamOutput out, Grok grok) throws IOException {
         out.writeLogicalPlanNode(grok.child());
         out.writeExpression(grok.input());
         out.writeString(grok.parser().pattern());
-        writeAttributes(out, grok.extractedFields());
+        // TODO: needs new transport version
+        writeAliases(out, grok.extractedFields());
     }
 
     static Limit readLimit(PlanStreamInput in) throws IOException {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizer.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.xpack.esql.expression.function.scalar.spatial.SpatialRelatesFunction;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.Equals;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.In;
+import org.elasticsearch.xpack.esql.plan.GeneratingPlan;
 import org.elasticsearch.xpack.esql.plan.logical.Aggregate;
 import org.elasticsearch.xpack.esql.plan.logical.Enrich;
 import org.elasticsearch.xpack.esql.plan.logical.EsRelation;
@@ -86,7 +87,6 @@
 import static java.util.Collections.singleton;
 import static org.elasticsearch.xpack.esql.expression.NamedExpressions.mergeOutputExpressions;
 import static org.elasticsearch.xpack.esql.optimizer.LogicalPlanOptimizer.SubstituteSurrogates.rawTemporaryName;
-import static org.elasticsearch.xpack.ql.expression.Expressions.asAttributes;
 import static org.elasticsearch.xpack.ql.optimizer.OptimizerRules.TransformDirection;
 import static org.elasticsearch.xpack.ql.optimizer.OptimizerRules.TransformDirection.DOWN;
 
@@ -961,21 +961,21 @@ private static LogicalPlan maybePushDownPastUnary(Filter filter, UnaryPlan unary
     protected static class PushDownEval extends OptimizerRules.OptimizerRule<Eval> {
         @Override
         protected LogicalPlan rule(Eval eval) {
-            return pushGeneratingPlanPastProjectAndOrderBy(eval, asAttributes(eval.fields()));
+            return pushGeneratingPlanPastProjectAndOrderBy(eval);
         }
     }
 
     protected static class PushDownRegexExtract extends OptimizerRules.OptimizerRule<RegexExtract> {
         @Override
         protected LogicalPlan rule(RegexExtract re) {
-            return pushGeneratingPlanPastProjectAndOrderBy(re, re.extractedFields());
+            return pushGeneratingPlanPastProjectAndOrderBy(re);
         }
     }
 
     protected static class PushDownEnrich extends OptimizerRules.OptimizerRule<Enrich> {
         @Override
         protected LogicalPlan rule(Enrich en) {
-            return pushGeneratingPlanPastProjectAndOrderBy(en, asAttributes(en.enrichFields()));
+            return pushGeneratingPlanPastProjectAndOrderBy(en);
         }
     }
 
@@ -1005,14 +1005,16 @@ protected LogicalPlan rule(Enrich en) {
      *
      * ... | eval $$a = a | eval a = b + 1 | sort $$a | drop $$a
      */
-    private static LogicalPlan pushGeneratingPlanPastProjectAndOrderBy(UnaryPlan generatingPlan, List<Attribute> generatedAttributes) {
+    private static <Plan extends UnaryPlan & GeneratingPlan<Plan>> LogicalPlan pushGeneratingPlanPastProjectAndOrderBy(
+        Plan generatingPlan
+    ) {
         LogicalPlan child = generatingPlan.child();
-
         if (child instanceof OrderBy orderBy) {
-            Set<String> evalFieldNames = new LinkedHashSet<>(Expressions.names(generatedAttributes));
+            // TODO: We can simplify this and make it similar to the pushdown past project by just renaming the generated attributes.
+            Set<String> generatedFieldNames = new LinkedHashSet<>(Expressions.names(generatingPlan.generatedAttributes()));
 
             // Look for attributes in the OrderBy's expressions and create aliases with temporary names for them.
-            AttributeReplacement nonShadowedOrders = renameAttributesInExpressions(evalFieldNames, orderBy.order());
+            AttributeReplacement nonShadowedOrders = renameAttributesInExpressions(generatedFieldNames, orderBy.order());
 
             AttributeMap<Alias> aliasesForShadowedOrderByAttrs = nonShadowedOrders.replacedAttributes;
             @SuppressWarnings("unchecked")
@@ -1030,9 +1032,66 @@ private static LogicalPlan pushGeneratingPlanPastProjectAndOrderBy(UnaryPlan gen
             }
 
             return orderBy.replaceChild(generatingPlan.replaceChild(orderBy.child()));
-        } else if (child instanceof Project) {
-            var projectWithEvalChild = pushDownPastProject(generatingPlan);
-            return projectWithEvalChild.withProjections(mergeOutputExpressions(generatedAttributes, projectWithEvalChild.projections()));
+        } else if (child instanceof Project project) {
+            AttributeMap.Builder<Expression> aliasBuilder = AttributeMap.builder();
+            project.forEachExpression(Alias.class, a -> aliasBuilder.put(a.toAttribute(), a.child()));
+            var aliases = aliasBuilder.build();
+
+            // Resolve Project's renames in the eval.
+            @SuppressWarnings("unchecked")
+            Plan generatingPlanWithResolvedExpressions = (Plan) generatingPlan.transformExpressionsOnly(
+                ReferenceAttribute.class,
+                r -> aliases.resolve(r, r)
+            );
+
+            // Look for generated Attributes that currently shadow any of the Project's references.
+            // We need to generate them using a different, non-shadowing name to avoid inconsistencies.
+            List<Attribute> generatedAttributes = generatingPlan.generatedAttributes();
+            Set<String> projectReferencedNames = project.references().names();
+            Map<String, String> renameGeneratedAttributeTo = new HashMap<>();
+            for (Attribute attr : generatedAttributes) {
+                String name = attr.name();
+                if (projectReferencedNames.contains(name)) {
+                    renameGeneratedAttributeTo.putIfAbsent(
+                        name,
+                        // TODO: Use e.g. AtomicLong to make sure generated temp names can not clash.
+                        // Do not use the attribute's id, as multiple attributes with the same name can occur.
+                        SubstituteSurrogates.rawTemporaryName(name, "temp_name", "")
+                    );
+                }
+            }
+            List<String> newNames = generatedAttributes.stream()
+                .map(attr -> renameGeneratedAttributeTo.getOrDefault(attr.name(), attr.name()))
+                .toList();
+            Plan generatingPlanWithRenamedAttributes = generatingPlanWithResolvedExpressions.withGeneratedNames(newNames);
+
+            // Put the project at the top, but include the generated attributes.
+            // Any generated attributes that had to be renamed need to be re-renamed to their original names.
+            List<NamedExpression> generatedAttributesRenamedToOriginal = new ArrayList<>(generatedAttributes.size());
+            List<Attribute> renamedGeneratedAttributes = generatingPlanWithRenamedAttributes.generatedAttributes();
+            for (int i = 0; i < generatedAttributes.size(); i++) {
+                Attribute originalAttribute = generatedAttributes.get(i);
+                Attribute renamedAttribute = renamedGeneratedAttributes.get(i);
+                if (originalAttribute.name().equals(renamedAttribute.name())) {
+                    generatedAttributesRenamedToOriginal.add(renamedAttribute);
+                } else {
+                    generatedAttributesRenamedToOriginal.add(
+                        new Alias(
+                            originalAttribute.source(),
+                            originalAttribute.name(),
+                            originalAttribute.qualifier(),
+                            renamedAttribute,
+                            originalAttribute.id(),
+                            originalAttribute.synthetic()
+                        )
+                    );
+                }
+            }
+
+            Project projectWithGeneratingChild = project.replaceChild(generatingPlanWithRenamedAttributes.replaceChild(project.child()));
+            return projectWithGeneratingChild.withProjections(
+                mergeOutputExpressions(generatedAttributesRenamedToOriginal, projectWithGeneratingChild.projections())
+            );
         }
 
         return generatingPlan;
@@ -1056,6 +1115,7 @@ private static AttributeReplacement renameAttributesInExpressions(
             rewrittenExpressions.add(expr.transformUp(Attribute.class, attr -> {
                 if (attributeNamesToRename.contains(attr.name())) {
                     Alias renamedAttribute = aliasesForReplacedAttributes.computeIfAbsent(attr, a -> {
+                        // TODO: Use e.g. AtomicLong to make sure generated temp names can not clash.
                         String tempName = SubstituteSurrogates.rawTemporaryName(a.name(), "temp_name", a.id().toString());
                         // TODO: this should be synthetic
                         return new Alias(a.source(), tempName, null, a, null, false);

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/OptimizerRules.java

@@ -104,6 +104,11 @@ protected AttributeSet references(LogicalPlan plan) {
                 // But they are not actually referring to attributes from the input plan - only the match field does.
                 return enrich.matchField().references();
             }
+            if (plan instanceof RegexExtract re) {
+                // Similarly as for Enrich: the extractedFields are Aliases for ReferenceAttributes, which are in turn created by the
+                // RegexExtract node.
+                return re.input().references();
+            }
             return super.references(plan);
         }
 
@@ -120,7 +125,7 @@ protected AttributeSet generates(LogicalPlan logicalPlan) {
                 return new AttributeSet(Expressions.asAttributes(eval.fields()));
             }
             if (logicalPlan instanceof RegexExtract extract) {
-                return new AttributeSet(extract.extractedFields());
+                return new AttributeSet(Expressions.asAttributes(extract.extractedFields()));
             }
             if (logicalPlan instanceof MvExpand mvExpand) {
                 return new AttributeSet(mvExpand.expanded());
@@ -156,7 +161,7 @@ protected AttributeSet generates(PhysicalPlan physicalPlan) {
                 return new AttributeSet(Expressions.asAttributes(eval.fields()));
             }
             if (physicalPlan instanceof RegexExtractExec extract) {
-                return new AttributeSet(extract.extractedFields());
+                return new AttributeSet(Expressions.asAttributes(extract.extractedFields()));
             }
             if (physicalPlan instanceof MvExpandExec mvExpand) {
                 return new AttributeSet(mvExpand.expanded());

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/PhysicalPlanOptimizer.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.xpack.esql.plan.physical.UnaryExec;
 import org.elasticsearch.xpack.ql.common.Failure;
 import org.elasticsearch.xpack.ql.expression.Alias;
+import org.elasticsearch.xpack.ql.expression.Attribute;
 import org.elasticsearch.xpack.ql.expression.AttributeMap;
 import org.elasticsearch.xpack.ql.expression.AttributeSet;
 import org.elasticsearch.xpack.ql.expression.Expression;
@@ -112,14 +113,16 @@ public PhysicalPlan apply(PhysicalPlan plan) {
                         }
                     });
                     if (p instanceof RegexExtractExec ree) {
-                        attributes.removeAll(ree.extractedFields());
+                        for (Alias extractedField : ree.extractedFields()) {
+                            attributes.remove((Attribute) extractedField.child());
+                        }
                     }
                     if (p instanceof MvExpandExec mvee) {
                         attributes.remove(mvee.expanded());
                     }
                     if (p instanceof EnrichExec ee) {
                         for (NamedExpression enrichField : ee.enrichFields()) {
-                            // TODO: why is this different then the remove above?
+                            // TODO: This weirdness could be avoided if Enrich just used Aliases all the time.
                             attributes.remove(enrichField instanceof Alias a ? a.child() : enrichField);
                         }
                     }

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

@@ -164,10 +164,10 @@ public PlanFactory visitDissectCommand(EsqlBaseParser.DissectCommandContext ctx)
                         referenceKeys.iterator().next()
                     );
                 }
-                List<Attribute> keys = new ArrayList<>();
+                List<Alias> keys = new ArrayList<>();
                 for (var x : parser.outputKeys()) {
                     if (x.isEmpty() == false) {
-                        keys.add(new ReferenceAttribute(src, x, DataTypes.KEYWORD));
+                        keys.add(new Alias(src, x, new ReferenceAttribute(src, x, DataTypes.KEYWORD)));
                     }
                 }
                 return new Dissect(src, p, expression(ctx.primaryExpression()), new Dissect.Parser(pattern, appendSeparator, parser), keys);

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.plan;
+
+import org.elasticsearch.xpack.ql.expression.Alias;
+import org.elasticsearch.xpack.ql.expression.Attribute;
+import org.elasticsearch.xpack.ql.expression.NameId;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public interface GeneratingPlan<PlanType extends GeneratingPlan<PlanType>> {
+    List<Attribute> generatedAttributes();
+
+    PlanType withGeneratedNames(List<String> newNames);
+
+    static List<Alias> renameAliases(List<Alias> originalAliases, List<String> newNames) {
+        if (newNames.size() != originalAliases.size()) {
+            throw new IllegalArgumentException(
+                "Number of new names is [" + newNames.size() + "] but there are [" + originalAliases.size() + "] names."
+            );
+        }
+
+        List<Alias> newFields = new ArrayList<>(originalAliases.size());
+        for (int i = 0; i < originalAliases.size(); i++) {
+            Alias field = originalAliases.get(i);
+            String newName = newNames.get(i);
+            if (field.name().equals(newName)) {
+                newFields.add(field);
+            } else {
+                newFields.add(new Alias(field.source(), newName, field.qualifier(), field.child(), new NameId(), field.synthetic()));
+            }
+        }
+
+        return newFields;
+    }
+}