refactor(esql): Improve MvExpand push-down logic in logical optimizer

- Refactored PushDownMvExpandPastProject to handle more complex projection scenarios
- Added support for preserving original projection aliases during MvExpand push-down
- Replaced previous push-down implementation with more robust alias resolution
- Ensured that output expressions maintain their original context and aliases
- Updated test case to reflect new push-down behavior
The changes improve the flexibility and accuracy of MvExpand push-down optimization in the logical plan optimizer.

Author: kanoshiou

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

@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.esql.optimizer.rules.logical;
 
 import org.elasticsearch.xpack.esql.core.expression.Alias;
+import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.AttributeMap;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.plan.logical.Eval;
@@ -68,8 +69,31 @@ protected LogicalPlan rule(MvExpand mvExpand) {
                 }
             }
 
-            Project project = PushDownUtils.pushDownPastProject(mvExpand);
-            return PushDownUtils.resolveRenamesFromMap(project, AttributeMap.of(mvExpand.target().toAttribute(), mvExpand.expanded()));
+            // 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));
+            AttributeMap<Alias> aliases = aliasBuilder.build();
+
+            // Push down the MvExpand past the Project
+            MvExpand pushedDownMvExpand = new MvExpand(mvExpand.source(), pj.child(), mvExpand.target(), mvExpand.expanded());
+
+            // 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);
+                if (alias == null) {
+                    // No alias, use the output expression directly
+                    newProjections.add(outputExpr);
+                } else if (alias.child().semanticEquals(mvExpand.target().toAttribute())) {
+                    // Alias child is the target attribute, replace it with the expanded attribute
+                    newProjections.add(alias.replaceChild(mvExpand.expanded()));
+                } else {
+                    // Keep the alias as is
+                    newProjections.add(alias);
+                }
+            }
+
+            return new Project(pj.source(), pushedDownMvExpand, newProjections);
         }
         return mvExpand;
     }

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

@@ -1670,11 +1670,11 @@ public void testCopyDefaultLimitPastLookupJoin() {
     /**
      * Expected
      * <pre>{@code
-     * EsqlProject[[first_name{r}#17, last_name{f}#10]]
-     * \_Limit[10[INTEGER],true]
-     *   \_MvExpand[first_name{f}#7,first_name{r}#17]
-     *     \_Limit[1[INTEGER],false]
-     *       \_EsRelation[test][_meta_field{f}#12, emp_no{f}#6, first_name{f}#7, ge..]
+     * Project[[first_name{r}#18, last_name{f}#11]]
+     * \_Limit[10[INTEGER],true,false]
+     *   \_MvExpand[first_name{f}#8,first_name{r}#18]
+     *     \_Limit[1[INTEGER],false,false]
+     *       \_EsRelation[test][_meta_field{f}#13, emp_no{f}#7, first_name{f}#8, ge..]
      * }</pre>
      */
     public void testDontPushDownLimitPastMvExpand() {
@@ -1685,7 +1685,7 @@ public void testDontPushDownLimitPastMvExpand() {
             | mv_expand first_name
             | limit 10
             """);
-        var project = as(plan, EsqlProject.class);
+        var project = as(plan, Project.class);
         var limit = asLimit(project.child(), 10, true);
         var mvExpand = as(limit.child(), MvExpand.class);
         var limit2 = asLimit(mvExpand.child(), 1, false);