stabilize tests

Author: bpintea

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

@@ -1656,16 +1656,16 @@ required_capability: join_lookup_v12
 
 FROM employees
 | RENAME languages AS language_code
-| SORT language_code
+| SORT emp_no, language_code
 | LIMIT 4
 | EVAL language_code = TO_INTEGER(NULL)
 | LOOKUP JOIN languages_lookup ON language_code
 | KEEP emp_no, language_code, language_name
 ;
 
 emp_no:integer | language_code:integer | language_name:keyword
-10009          |null           |null
-10013          |null           |null
-10019          |null           |null
-10005          |null           |null
+10001          |null           |null
+10002          |null           |null
+10003          |null           |null
+10004          |null           |null
 ;

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

@@ -38,31 +38,32 @@ public class PruneJoinOnNullMatchingField extends OptimizerRules.OptimizerRule<J
 
     @Override
     protected LogicalPlan rule(Join join) {
+        LogicalPlan plan = join;
         var joinType = join.config().type();
-        if ((joinType == INNER || joinType == LEFT) == false) { // other types will have different replacement logic
-            return join;
-        }
-        AttributeMap.Builder<Expression> attributeMapBuilder = AttributeMap.builder();
-        loop: for (var child = join.left();; child = ((UnaryPlan) child).child()) { // cast is safe as both Project and Eval are UnaryPlans
-            switch (child) {
-                case Project project -> project.projections().forEach(projection -> {
-                    if (projection instanceof Alias alias) {
-                        attributeMapBuilder.put(alias.toAttribute(), alias.child());
+        if (joinType == INNER || joinType == LEFT) { // other types will have different replacement logic
+            AttributeMap.Builder<Expression> attributeMapBuilder = AttributeMap.builder();
+            loop: for (var child = join.left();; child = ((UnaryPlan) child).child()) { // cast is safe as both plans are UnaryPlans
+                switch (child) {
+                    case Project project -> project.projections().forEach(projection -> {
+                        if (projection instanceof Alias alias) {
+                            attributeMapBuilder.put(alias.toAttribute(), alias.child());
+                        }
+                    });
+                    case Eval eval -> eval.fields().forEach(alias -> attributeMapBuilder.put(alias.toAttribute(), alias.child()));
+                    default -> {
+                        break loop;
                     }
-                });
-                case Eval eval -> eval.fields().forEach(alias -> attributeMapBuilder.put(alias.toAttribute(), alias.child()));
-                default -> {
-                    break loop;
                 }
             }
-        }
-        for (var attr : AttributeSet.of(join.config().matchFields())) {
-            var resolved = attributeMapBuilder.build().resolve(attr);
-            if (resolved != null && isGuaranteedNull(resolved)) {
-                return replaceJoin(join);
+            for (var attr : AttributeSet.of(join.config().matchFields())) {
+                var resolved = attributeMapBuilder.build().resolve(attr);
+                if (resolved != null && isGuaranteedNull(resolved)) {
+                    plan = replaceJoin(join);
+                    break;
+                }
             }
         }
-        return join;
+        return plan;
     }
 
     private static LogicalPlan replaceJoin(Join join) {

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

@@ -7666,7 +7666,7 @@ private LocalExecutionPlanner.LocalExecutionPlan physicalOperationsFromPhysicalP
         // The TopN needs an estimated row size for the planner to work
         var plans = PlannerUtils.breakPlanBetweenCoordinatorAndDataNode(EstimatesRowSize.estimateRowSize(0, plan), config);
         plan = useDataNodePlan ? plans.v2() : plans.v1();
-        plan = PlannerUtils.localPlan(List.of(), config, FoldContext.small(), plan);
+        plan = PlannerUtils.localPlan(config, FoldContext.small(), plan, TEST_SEARCH_STATS);
         ExchangeSinkHandler exchangeSinkHandler = new ExchangeSinkHandler(null, 10, () -> 10);
         LocalExecutionPlanner planner = new LocalExecutionPlanner(
             "test",