Fix tests

Author: ioanatia

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

@@ -161,7 +161,7 @@ fork1         | 10081          | null   | null      | null
 fork2         | 10081          | null   | null      | null
 fork2         | 10087          | null   | null      | null
 fork3         | null           | 100    | 10100     | 10001
-fork4         | null           | null   | null      | null
+fork4         | null           | 100    | 10001     | null
 ;
 
 forkWithDrop-Ignore
@@ -223,8 +223,8 @@ FROM employees
 _fork:keyword | emp_no:integer | x:keyword | y:keyword | z:keyword | w:keyword
 fork1         | 10048          | Florian   | 10048     | Syrotiuk  | null
 fork1         | 10081          | Zhongwei  | 10081     | Rosen     | null
-fork2         | 10048          | null      | null      | null      | Florian
-fork2         | 10081          | null      | null      | null      | Zhongwei
+fork2         | 10048          | Syrotiuk  | 10048     | null      | Florian
+fork2         | 10081          | Rosen     | 10081     | null      | Zhongwei
 ;
 
 forkWithMixOfCommands

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/ForkIT.java

@@ -521,6 +521,7 @@ public void testWithEvalDifferentOutputs() {
                 | FORK ( EVAL a = 1 )
                        ( EVAL b = 2 )
                 | KEEP a, b, _fork
+                | SORT _fork, a
             """;
         try (var resp = run(query)) {
             assertColumnNames(resp.columns(), List.of("a", "b", "_fork"));
@@ -533,6 +534,7 @@ public void testWithStatsSimple() {
                 | FORK (STATS x=COUNT(*), y=VALUES(id))
                        (WHERE id == 2)
                 | KEEP _fork, x, y, id
+                | SORT _fork, id
             """;
         try (var resp = run(query)) {
             assertColumnNames(resp.columns(), List.of("_fork", "x", "y", "id"));

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

@@ -706,7 +706,10 @@ private LogicalPlan resolveFork(Fork fork, AnalyzerContext context) {
                 }
 
                 List<String> subPlanColumns = logicalPlan.output().stream().map(Attribute::name).collect(Collectors.toList());
-                if (subPlanColumns.equals(forkColumns) == false) {
+                // We need to add an explicit Keep even if the outputs align
+                // This is because at the moment the sub plans are executed and optimized separately and the output might change
+                // during optimizations. Once we add streaming we might not need to add a Keep when the outputs already align.
+                if (logicalPlan instanceof Keep == false || subPlanColumns.equals(forkColumns) == false) {
                     changed = true;
                     List<Attribute> newOutput = new ArrayList<>();
                     for (String attrName : forkColumns) {