WIP

Author: julian-elastic

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

@@ -14,10 +14,13 @@
 import org.elasticsearch.xpack.esql.optimizer.rules.PlanConsistencyChecker;
 import org.elasticsearch.xpack.esql.plan.logical.Enrich;
 import org.elasticsearch.xpack.esql.plan.physical.EnrichExec;
+import org.elasticsearch.xpack.esql.plan.physical.EsQueryExec;
 import org.elasticsearch.xpack.esql.plan.physical.FieldExtractExec;
 import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
 
+import static org.elasticsearch.index.IndexMode.LOOKUP;
 import static org.elasticsearch.xpack.esql.common.Failure.fail;
+import static org.elasticsearch.xpack.esql.core.expression.Attribute.semanticEquals;
 
 /** Physical plan verifier. */
 public final class PhysicalVerifier {
@@ -67,10 +70,12 @@ public Failures verify(PhysicalPlan planAfter, boolean skipRemoteEnrichVerificat
             }
         });
 
-        if (planBefore.output().equals(planAfter.output()) == false) {
-            failures.add(
-                fail(planAfter, "Layout has changed from [{}] to [{}]. ", planBefore.output().toString(), planAfter.output().toString())
-            );
+        if (semanticEquals(planBefore.output(), planAfter.output()) == false) {
+            if ((planAfter instanceof EsQueryExec esQueryExec && esQueryExec.indexMode() == LOOKUP) == false) {
+                failures.add(
+                    fail(planAfter, "Layout has changed from [{}] to [{}]. ", planBefore.output().toString(), planAfter.output().toString())
+                );
+            }
         }
 
         if (depFailures.hasFailures()) {

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

@@ -185,7 +185,15 @@ public static PhysicalPlan localPlan(
     ) {
         final LocalMapper localMapper = new LocalMapper();
         var isCoordPlan = new Holder<>(Boolean.TRUE);
-
+        /*Set<FragmentExec> fragmentExecsUnderLookupJoin = new HashSet<>();
+        plan.forEachDown(LookupJoinExec.class, lookupJoin -> {
+            ;
+            // If the LookupJoinExec is the parent of a FragmentExec, we need to skip it
+            if (lookupJoin.right() instanceof FragmentExec fragmentExec) {
+                // this change is disabled for now, as it seems physicalOptimizer.localOptimize is still needed even for lookup joins
+                // fragmentExecsUnderLookupJoin.add(fragmentExec);
+            }
+        });*/
         var localPhysicalPlan = plan.transformUp(FragmentExec.class, f -> {
             isCoordPlan.set(Boolean.FALSE);
             var optimizedFragment = logicalOptimizer.localOptimize(f.fragment());
@@ -204,7 +212,12 @@ public static PhysicalPlan localPlan(
                     )
                 );
             }
-            var localOptimized = physicalOptimizer.localOptimize(physicalFragment);
+            PhysicalPlan localOptimized = physicalOptimizer.localOptimize(physicalFragment);
+            /*if (fragmentExecsUnderLookupJoin.contains(f)) {
+                localOptimized = physicalFragment;
+            } else {
+                localOptimized = physicalOptimizer.localOptimize(physicalFragment);
+            }*/
             return EstimatesRowSize.estimateRowSize(f.estimatedRowSize(), localOptimized);
         });
         return isCoordPlan.get() ? plan : localPhysicalPlan;

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

@@ -175,7 +175,7 @@ protected final ExecutionInfo executeWithInfo(TreeType plan) {
                     if (tf.hasChanged()) {
                         hasChanged = true;
                         if (log.isTraceEnabled()) {
-                            log.trace("Rule {} applied\n{}", rule, NodeUtils.diffString(tf.before, tf.after));
+                            log.trace("Rule {} applied with change\n{}", rule, NodeUtils.diffString(tf.before, tf.after));
                         }
                     } else {
                         if (log.isTraceEnabled()) {

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

@@ -31,6 +31,7 @@
 import org.elasticsearch.search.vectors.KnnVectorQueryBuilder;
 import org.elasticsearch.search.vectors.RescoreVectorBuilder;
 import org.elasticsearch.test.VersionUtils;
+import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.xpack.core.enrich.EnrichPolicy;
 import org.elasticsearch.xpack.esql.EsqlTestUtils;
 import org.elasticsearch.xpack.esql.EsqlTestUtils.TestSearchStats;
@@ -146,7 +147,7 @@
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
 
-//@TestLogging(value = "org.elasticsearch.xpack.esql:TRACE,org.elasticsearch.compute:TRACE", reason = "debug")
+@TestLogging(value = "org.elasticsearch.xpack.esql:TRACE,org.elasticsearch.compute:TRACE", reason = "debug")
 public class LocalPhysicalPlanOptimizerTests extends MapperServiceTestCase {
 
     public static final List<DataType> UNNECESSARY_CASTING_DATA_TYPES = List.of(