Remove limit optimization - data node optimizer will take care of it

Author: smalyshev

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

@@ -126,13 +126,7 @@ else if (aggregate.groupings()
         }
 
         if (unary instanceof Limit limit) {
-            // Avoid extra limit for remote Enrich, since it has already been applied before Enrich if it's duplicate.
-            // TODO: can we do this in the optimizer instead?
-            if (limit.duplicated() && limit.child() instanceof Enrich) {
-                mappedChild = addExchangeForFragment(limit.child(), mappedChild);
-            } else {
-                mappedChild = addExchangeForFragment(limit, mappedChild);
-            }
+            mappedChild = addExchangeForFragment(limit, mappedChild);
             return new LimitExec(limit.source(), mappedChild, limit.limit(), null);
         }
 

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

@@ -7017,11 +7017,15 @@ public void testEnrichBeforeLimit() {
             var finalLimit = as(plan, LimitExec.class);
             var exchange = as(finalLimit.child(), ExchangeExec.class);
             var fragment = as(exchange.child(), FragmentExec.class);
-            var enrich = as(fragment.fragment(), Enrich.class);
+            var enrichLimit = as(fragment.fragment(), Limit.class);
+            assertThat(Foldables.limitValue(enrichLimit.limit(), enrichLimit.sourceText()), equalTo(10));
+            var enrich = as(enrichLimit.child(), Enrich.class);
             assertThat(enrich.mode(), equalTo(Enrich.Mode.REMOTE));
             assertThat(enrich.concreteIndices(), equalTo(Map.of("cluster_1", ".enrich-departments-2")));
             var evalFragment = as(enrich.child(), Eval.class);
             var partialLimit = as(evalFragment.child(), Limit.class);
+            assertThat(Foldables.limitValue(partialLimit.limit(), partialLimit.sourceText()), equalTo(10));
+            assertTrue(partialLimit.local());
             as(partialLimit.child(), EsRelation.class);
         }
     }
@@ -7075,11 +7079,15 @@ public void testLimitThenEnrichRemote() {
         var finalLimit = as(plan, LimitExec.class);
         var exchange = as(finalLimit.child(), ExchangeExec.class);
         var fragment = as(exchange.child(), FragmentExec.class);
-        var enrich = as(fragment.fragment(), Enrich.class);
+        var enrichLimit = as(fragment.fragment(), Limit.class);
+        assertThat(Foldables.limitValue(enrichLimit.limit(), enrichLimit.sourceText()), equalTo(10));
+        var enrich = as(enrichLimit.child(), Enrich.class);
         assertThat(enrich.mode(), equalTo(Enrich.Mode.REMOTE));
         assertThat(enrich.concreteIndices(), equalTo(Map.of("cluster_1", ".enrich-departments-2")));
         var evalFragment = as(enrich.child(), Eval.class);
         var partialLimit = as(evalFragment.child(), Limit.class);
+        assertThat(Foldables.limitValue(enrichLimit.limit(), enrichLimit.sourceText()), equalTo(10));
+        assertTrue(partialLimit.local());
         as(partialLimit.child(), EsRelation.class);
     }