fix: refine PushDownMvExpandPastProject logic by correcting the alias collection and improving the pushdown skip condition.

Author: kanoshiou

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PushDownMvExpandPastProject.java

@@ -26,15 +26,24 @@ public final class PushDownMvExpandPastProject extends OptimizerRules.OptimizerR
     protected LogicalPlan rule(MvExpand mvExpand) {
         if (mvExpand.child() instanceof Project pj) {
             List<NamedExpression> projections = new ArrayList<>(pj.projections());
+            List<Attribute> output = mvExpand.output();
 
-            // Skip if any projection alias shadows an input field name, as injecting an Eval would cause
-            // duplicate output attributes. This can happen with aliases generated by ResolveUnionTypesInUnionAll.
+            // Skip if the expanded field has the same name as a field in the projection's input set, and
+            // the projection shadows that specific field from the projection input set.
+            // Pushing down the MvExpand in such cases would cause duplicate output attributes.
+            // This can happen with aliases generated by ResolveUnionTypesInUnionAll.
             // Example:
             // MvExpand[salary{r}#168,salary{r}#175]
             // \_Project[[$$salary$converted_to$keyword{r$}#178 AS salary#168]]
             //   \_UnionAll[[salary{r}#174, $$salary$converted_to$keyword{r$}#178]]
+            String expandedFieldName = mvExpand.expanded().name();
             Set<String> inputNames = pj.inputSet().stream().map(NamedExpression::name).collect(Collectors.toSet());
-            if (projections.stream().anyMatch(e -> e instanceof Alias alias && inputNames.contains(alias.toAttribute().name()))) {
+            if (projections.stream()
+                .anyMatch(
+                    e -> e instanceof Alias alias
+                        && inputNames.contains(expandedFieldName)
+                        && inputNames.contains(alias.toAttribute().name())
+                )) {
                 return mvExpand;
             }
 
@@ -56,7 +65,7 @@ protected LogicalPlan rule(MvExpand mvExpand) {
                             break;
                         }
 
-                        // for query like: row a = 2 | eval b = a ｜ keep * | mv_expand b
+                        // for query like: row a = 2 | eval b = a | keep * | mv_expand b
                         Alias aliasAlias = new Alias(
                             alias.source(),
                             TemporaryNameUtils.temporaryName(alias.child(), alias.toAttribute(), 0),
@@ -67,7 +76,7 @@ protected LogicalPlan rule(MvExpand mvExpand) {
                         mvExpand = new MvExpand(mvExpand.source(), pj, aliasAlias.toAttribute(), mvExpand.expanded());
                         break;
                     } else if (alias.child().semanticEquals(mvExpand.target().toAttribute())) {
-                        // for query like: row a = 2 | eval b = a ｜ keep * | mv_expand a
+                        // for query like: row a = 2 | eval b = a | keep * | mv_expand a
                         Alias aliasAlias = new Alias(
                             alias.source(),
                             TemporaryNameUtils.temporaryName(alias.child(), alias.toAttribute(), 0),
@@ -83,16 +92,20 @@ protected LogicalPlan rule(MvExpand mvExpand) {
 
             // Build a map of aliases from the original projection
             AttributeMap.Builder<Alias> aliasBuilder = AttributeMap.builder();
-            pj.forEachExpression(Alias.class, alias -> aliasBuilder.put(alias.toAttribute(), alias));
+            projections.forEach(ne -> {
+                if (ne instanceof Alias alias) {
+                    aliasBuilder.put(alias.toAttribute(), alias);
+                }
+            });
             AttributeMap<Alias> aliases = aliasBuilder.build();
 
             // Push down the MvExpand past the Project
             MvExpand pushedDownMvExpand = mvExpand.replaceChild(pj.child());
 
             // Create a new projection at the top based on mvExpand.output(), plugging back in the aliases
             List<NamedExpression> newProjections = new ArrayList<>();
-            for (Attribute outputExpr : mvExpand.output()) {
-                Alias alias = aliases.resolve(outputExpr, null);
+            for (Attribute outputExpr : output) {
+                Alias alias = aliases.get(outputExpr);
                 if (alias == null) {
                     // No alias, use the output expression directly
                     newProjections.add(outputExpr);