Fix PushDownAndCombineLimits local limit handling when the limits are equal (elastic#139399) (elastic#139477)

* Fix PushDownAndCombineLimits local limit handling when the limits are equal

(cherry picked from commit c80fa9a)

Author: smalyshev

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

@@ -83,9 +83,20 @@ private static Limit combineLimits(Limit upper, Limit lower, FoldContext ctx) {
         // Keep the smallest limit
         var upperLimitValue = (int) upper.limit().fold(ctx);
         var lowerLimitValue = (int) lower.limit().fold(ctx);
-        // We want to preserve the duplicated() value of the smaller limit.
-        if (lowerLimitValue <= upperLimitValue) {
-            return lower.withLocal(lower.local());
+        /*
+         * We always want to select the smaller of the limits, but with the local flag it gets a bit tricky.
+         * If one of the limits is smaller, that's what we will choose. But if the limits are exactly equal,
+         * then we can choose the local limit because this may enable some queries that would otherwise be prohibited
+         * due to pipeline-breaking nature of non-local limits in combination with remote Enrich.
+         * However this may not be true if we have more situations where local limits are generated which do not have
+         * guarantees that local limit produced by remote enrich pushing provides.
+         * See also: https://github.com/elastic/elasticsearch/pull/139399#pullrequestreview-3573026118
+         */
+        if (lowerLimitValue < upperLimitValue) {
+            return lower;
+        } else if (lowerLimitValue == upperLimitValue) {
+            // If any of them is local, we want the local limit
+            return lower.local() ? lower : lower.withLocal(upper.local());
         } else {
             return new Limit(upper.source(), upper.limit(), lower.child(), upper.duplicated(), upper.local());
         }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PushDownAndCombineLimitsTests.java

@@ -242,7 +242,8 @@ public void testPushableLimitDuplicate() {
             int precedingLimitValue = randomIntBetween(1, 10_000);
             Limit precedingLimit = new Limit(EMPTY, new Literal(EMPTY, precedingLimitValue, INTEGER), relation);
             LogicalPlan duplicatingLimitTestPlan = duplicatingTestCase.buildPlan(precedingLimit, a);
-            int upperLimitValue = randomIntBetween(1, precedingLimitValue);
+            // Explicitly raise the probability of equal limits, to test for https://github.com/elastic/elasticsearch/issues/139250
+            int upperLimitValue = randomBoolean() ? precedingLimitValue : randomIntBetween(1, precedingLimitValue);
             Limit upperLimit = new Limit(EMPTY, new Literal(EMPTY, upperLimitValue, INTEGER), duplicatingLimitTestPlan);
             Limit optimizedPlan = as(optimizePlan(upperLimit), Limit.class);
             duplicatingTestCase.checkOptimizedPlan(duplicatingLimitTestPlan, optimizedPlan.child());