ES|QL: Fix layout management for Project (#102399)

Fixes https://github.com/elastic/elasticsearch/issues/102120

In case of multiple aggs with the same expression, the layout has
multiple entries pointing to the same position. Project planning didn't
manage this situation properly, resulting in the same entry appearing
multiple times in the layout and pointing to multiple positions (that
violates the layout contract). The planning of the following commands
(eg. MV_EXPAND) resulted in a NullPointerException, because the
layout.inverse() could not be calculated properly.

docs/changelog/102399.yaml

@@ -0,0 +1,6 @@
+pr: 102399
+summary: "ES|QL: Fix layout management for Project"
+area: ES|QL
+type: bug
+issues:
+ - 102120

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

@@ -252,3 +252,57 @@ emp_no:integer  | first_name:keyword | generate_mv:keyword
 10001           | Georgi             | foo
 10002           | Bezalel            | foo
 ;
+
+
+// see https://github.com/elastic/elasticsearch/issues/102120
+expandAfterDuplicateAggs#[skip:-8.11.99]
+row a = 1 |  stats  a = count(*), b = count(*) | mv_expand b;
+
+a:long | b:long 
+1      | 1
+;
+
+expandAfterDuplicateAggs2#[skip:-8.11.99]
+row a = 1 | stats a = count(*), b = count(*) | mv_expand a;
+
+a:long | b:long 
+1      | 1
+;
+
+
+expandAfterDuplicateAggsAndEval#[skip:-8.11.99]
+row a = 1 | stats  a = count(*), b = count(*) | eval c = 2 | mv_expand b;
+
+a:long | b:long | c:integer
+1      | 1      | 2
+;
+
+expandAfterDuplicateAggsComplex#[skip:-8.11.99]
+row x = [1, 2, 3] 
+| mv_expand x 
+| stats a = count(*), b = count(*), c = count(*) 
+| eval x  = a + c + b 
+| mv_expand a 
+| rename a AS a | drop a;
+
+b:long    | c:long    | x:long
+3         | 3         | 9
+;
+
+
+expandAfterDuplicateAggsMultirow#[skip:-8.11.99]
+from employees 
+| stats a = count(gender), b = count(*), c = count(*) by gender 
+| eval str = concat(to_string(b), ",", gender) 
+| mv_expand b 
+| eval x = split(str,",") 
+| mv_expand x 
+| sort x;
+
+a:long      | b:long      | c:long      | gender:keyword | str:keyword   | x:keyword       
+33          |33           |33           |F               |"33,F"         |33             
+57          |57           |57           |M               |"57,M"         |57             
+33          |33           |33           |F               |"33,F"         |F              
+57          |57           |57           |M               |"57,M"         |M              
+0           |10           |10           |null            |null         |null           
+;

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

@@ -197,3 +197,12 @@ is_infinite              |? is_infinite(arg1:?)
 is_nan                   |? is_nan(arg1:?)                                        |arg1                     |?                 |  ""                                                  |?              | ""                      | false                | false
 // end::showFunctionsFiltered-result[]
 ;
+
+
+// see https://github.com/elastic/elasticsearch/issues/102120
+countFunctions#[skip:-8.11.99]
+show functions |  stats  a = count(*), b = count(*), c = count(*) |  mv_expand c;
+
+a:long | b:long | c:long
+82     | 82     | 82
+;

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java

@@ -585,16 +585,17 @@ public class LocalExecutionPlanner {
                 inputId = ne.id();
             }
             Layout.ChannelAndType input = source.layout.get(inputId);
-            Layout.ChannelSet channelSet = inputChannelToOutputIds.computeIfAbsent(
-                input.channel(),
-                ignore -> new Layout.ChannelSet(new HashSet<>(), input.type())
-            );
+            Layout.ChannelSet channelSet = inputChannelToOutputIds.get(input.channel());
+            if (channelSet == null) {
+                channelSet = new Layout.ChannelSet(new HashSet<>(), input.type());
+                channelSet.nameIds().add(ne.id());
+                layout.append(channelSet);
+            } else {
+                channelSet.nameIds().add(ne.id());
+            }
             if (channelSet.type() != input.type()) {
                 throw new IllegalArgumentException("type mismatch for aliases");
             }
-            channelSet.nameIds().add(ne.id());
-
-            layout.append(channelSet);
             projectionList.add(input.channel());
         }