Address more code review feedback

Author: julian-elastic

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

@@ -160,7 +160,9 @@ private static LogicalPlan pushDownPastJoin(Filter filter, Join join, FoldContex
                 optimizationApplied = true;
             }
             // push the right scoped filter down to the right child
-            // We don't want to execute this rule more than once per join, so we check if we
+            // We check if each AND component of the filter is already part of the right side filter before we add it
+            // In the future, this optimization can apply to other types of joins as well such as InlineJoin
+            // but for now we limit it to LEFT joins only, till filters are supported for other join types
             if (scoped.rightFilters().isEmpty() == false) {
                 List<Expression> rightPushableFilters = buildRightPushableFilters(scoped.rightFilters(), foldCtx);
                 if (rightPushableFilters.isEmpty() == false) {
@@ -220,6 +222,12 @@ private static List<Expression> buildRightPushableFilters(List<Expression> expre
     /**
      * Determines if the given expression can be pushed down to the right side of a join.
      * A filter is right pushable if the filter's predicate evaluates to false or null when all fields are set to null
+     * This rule helps us guard against the case where we don't know if a field is null because:
+     * 1. the field is null in the source data or
+     * 2. the field is null because there was no match in the join
+     * If the null could be an issue we just say the filter is not pushable and we avoid this issue
+     * In this context pushable means that we can push the filter down to the right side of a LEFT join
+     * We do not check if the filter is pushable to Lucene or not here
      */
     private static boolean isRightPushableFilter(Expression filter, FoldContext foldCtx) {
         // traverse the filter tree

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

@@ -95,7 +95,6 @@
 import org.elasticsearch.xpack.esql.plan.logical.EsRelation;
 import org.elasticsearch.xpack.esql.plan.logical.Fork;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
-import org.elasticsearch.xpack.esql.plan.logical.UnaryPlan;
 import org.elasticsearch.xpack.esql.plan.physical.AggregateExec;
 import org.elasticsearch.xpack.esql.plan.physical.ChangePointExec;
 import org.elasticsearch.xpack.esql.plan.physical.DissectExec;
@@ -744,7 +743,7 @@ private PhysicalOperation planLookupJoin(LookupJoinExec join, LocalExecutionPlan
         }
         Layout layout = layoutBuilder.build();
 
-        EsRelation esRelation = fildEsRelation(join.lookup());
+        EsRelation esRelation = findEsRelation(join.lookup());
         if (esRelation == null || esRelation.indexMode() != IndexMode.LOOKUP) {
             throw new IllegalArgumentException("can't plan [" + join + "]");
         }
@@ -809,18 +808,12 @@ private PhysicalOperation planLookupJoin(LookupJoinExec join, LocalExecutionPlan
         );
     }
 
-    private EsRelation fildEsRelation(PhysicalPlan node) {
+    private static EsRelation findEsRelation(PhysicalPlan node) {
         if (node instanceof FragmentExec fragmentExec) {
-            return fildEsRelation(fragmentExec.fragment());
-        }
-        return null;
-    }
-
-    private EsRelation fildEsRelation(LogicalPlan node) {
-        if (node instanceof EsRelation esRelation) {
-            return esRelation;
-        } else if (node instanceof UnaryPlan unaryPlan) {
-            return fildEsRelation(unaryPlan.child());
+            List<LogicalPlan> esRelations = fragmentExec.fragment().collectFirstChildren(x -> x instanceof EsRelation);
+            if (esRelations.size() == 1) {
+                return (EsRelation) esRelations.get(0);
+            }
         }
         return null;
     }

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

@@ -216,7 +216,9 @@ public static PhysicalPlan localPlan(
 
         var localPhysicalPlan = plan.transformUp(FragmentExec.class, f -> {
             if (lookupJoinExecRightChildren.contains(f)) {
-                // do not optimize the right child of a lookup join exec
+                // Do not optimize the right child of a lookup join exec
+                // The data node does not have the right stats to perform the optimization because the stats are on the lookup node
+                // Also we only ship logical plans across the network, so the plan needs to remain logical
                 return f;
             }
             isCoordPlan.set(Boolean.FALSE);

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

@@ -228,11 +228,7 @@ private PhysicalPlan mapBinary(BinaryPlan bp) {
                 );
             }
             if (right instanceof FragmentExec fragment) {
-                boolean isIndexModeLookup = fragment.fragment() instanceof EsRelation relation && relation.indexMode() == IndexMode.LOOKUP;
-                isIndexModeLookup = isIndexModeLookup
-                    || fragment.fragment() instanceof Filter filter
-                        && filter.child() instanceof EsRelation relation
-                        && relation.indexMode() == IndexMode.LOOKUP;
+                boolean isIndexModeLookup = isIndexModeLookup(fragment);
                 if (isIndexModeLookup) {
                     return new LookupJoinExec(
                         join.source(),
@@ -248,6 +244,18 @@ private PhysicalPlan mapBinary(BinaryPlan bp) {
         return MapperUtils.unsupported(bp);
     }
 
+    private static boolean isIndexModeLookup(FragmentExec fragment) {
+        // we support 2 cases:
+        // EsRelation in index_mode=lookup
+        boolean isIndexModeLookup = fragment.fragment() instanceof EsRelation relation && relation.indexMode() == IndexMode.LOOKUP;
+        // or Filter(EsRelation) in index_mode=lookup
+        isIndexModeLookup = isIndexModeLookup
+            || fragment.fragment() instanceof Filter filter
+                && filter.child() instanceof EsRelation relation
+                && relation.indexMode() == IndexMode.LOOKUP;
+        return isIndexModeLookup;
+    }
+
     private PhysicalPlan mapFork(Fork fork) {
         return new MergeExec(fork.source(), fork.children().stream().map(child -> map(child)).toList(), fork.output());
     }

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

@@ -41,7 +41,7 @@ public class EsqlFlags {
         Setting.Property.NodeScope,
         Setting.Property.Dynamic
     );
-
+    public static List<Setting<?>> allEsqlFlagsSettings = List.of(ESQL_STRING_LIKE_ON_INDEX, ESQL_ROUNDTO_PUSHDOWN_THRESHOLD);
     private final boolean stringLikeOnIndex;
 
     private final int roundToPushdownThreshold;
@@ -84,6 +84,6 @@ public int roundToPushdownThreshold() {
     }
 
     public static List<Setting<?>> listAll() {
-        return List.of(ESQL_STRING_LIKE_ON_INDEX, ESQL_ROUNDTO_PUSHDOWN_THRESHOLD);
+        return allEsqlFlagsSettings;
     }
 }