Address more code review comments

Author: julian-elastic

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

@@ -133,17 +133,17 @@ protected LookupEnrichQueryGenerator queryList(
     /**
      * This function will perform any planning needed on the local node
      * For now, we will just do mapping of the logical plan to physical plan
-     * In the future we can also do local physical and logical optimizations
+     * In the future we can also do local physical and logical optimizations.
+     * We only support a FragmentExec node containing a logical plan or a null plan
+     * If any other plan is sent we will just return null. This can happen in cases
+     * where the coordinator is running an older version that does not support
+     * keeping the plan as Logical Plan inside FragmentExec yet
+     * In those cases, it is safe to ignore the plan sent and return null
      */
     private static PhysicalPlan localLookupNodePlanning(PhysicalPlan physicalPlan) {
         if (physicalPlan instanceof FragmentExec fragmentExec) {
-            try {
-                LocalMapper localMapper = new LocalMapper();
-                return localMapper.map(fragmentExec.fragment());
-            } catch (Exception e) {
-                logger.error(() -> "Failed to perform local mapping on the lookup node for plan: [" + physicalPlan + "]", e);
-                return null;
-            }
+            LocalMapper localMapper = new LocalMapper();
+            return localMapper.map(fragmentExec.fragment());
         }
         return null;
     }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/BinaryExec.java

@@ -49,11 +49,7 @@ public PhysicalPlan right() {
     public void writeTo(StreamOutput out) throws IOException {
         Source.EMPTY.writeTo(out);
         out.writeNamedWriteable(left);
-        out.writeNamedWriteable(getRightToSerialize(out));
-    }
-
-    protected PhysicalPlan getRightToSerialize(StreamOutput out) {
-        return right;
+        out.writeNamedWriteable(right);
     }
 
     @Override

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/LookupJoinExec.java

@@ -7,7 +7,6 @@
 
 package org.elasticsearch.xpack.esql.plan.physical;
 
-import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -69,20 +68,6 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeNamedWriteableCollection(addedFields);
     }
 
-    @Override
-    protected PhysicalPlan getRightToSerialize(StreamOutput out) {
-        PhysicalPlan rightToSerialize = right();
-        if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_LOOKUP_JOIN_PRE_JOIN_FILTER) == false) {
-            // Prior to TransportVersions.ESQL_LOOKUP_JOIN_PRE_JOIN_FILTER
-            // we do not support a filter on top of the right side of the join
-            // As we consider the filters optional, we remove them here
-            while (rightToSerialize instanceof FilterExec filterExec) {
-                rightToSerialize = filterExec.child();
-            }
-        }
-        return rightToSerialize;
-    }
-
     @Override
     public String getWriteableName() {
         return ENTRY.name;