ESQL: Fix lookup join filter pushdown to use semantic equality (#136818) (#136999)

Previously, when pushing down filters to the right side of a lookup
join, we used simple equality (contains) to check for duplicate filters.
This could result in semantically equivalent filters being added
multiple times if they were different object instances.

This commit changes the duplicate check to use `semanticEquals()`
instead, ensuring that filters with the same semantic meaning are
properly deduplicated regardless of object identity.

Closes #136599

(cherry picked from commit f54987f)

# Conflicts:
#	x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/generative/GenerativeRestTest.java
#	x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

Co-authored-by: kanoshiou <[email protected]>

Author: alex-spies

docs/changelog/136818.yaml

@@ -0,0 +1,6 @@
+pr: 136818
+summary: "ESQL: Fix lookup join filter pushdown to use semantic equality"
+area: ES|QL
+type: bug
+issues:
+  - 136599

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClusterSpecIT.java

@@ -119,6 +119,7 @@ public MultiClusterSpecIT(
         // Lookup join after LIMIT is not supported in CCS yet
         "LookupJoinAfterLimitAndRemoteEnrich",
         "LookupJoinExpressionAfterLimitAndRemoteEnrich",
+        "LookupJoinWithSemanticFilterDeduplicationComplex",
         // Lookup join after FORK is not support in CCS yet
         "ForkBeforeLookupJoin"
     );

x-pack/plugin/esql/qa/testFixtures/src/main/resources/lookup-join.csv-spec

@@ -5647,3 +5647,46 @@ from sample_data,languages_mixed_numerics,apps
 null           |null           |null           |null           |32767.0               |32767.0               |32768.0                 |32767                |32767               |32767.0                   |32767              |null                 |null           |null           |null
 null           |null           |null           |null           |128.0                 |128.0                 |128.0                   |128                  |128                 |128.0                     |128                |null                 |null           |null           |null
 ;
+
+
+lookupJoinWithSemanticFilterDeduplication
+required_capability: join_lookup_v12
+required_capability: lookup_join_semantic_filter_dedup
+
+from languages
+| lookup join languages_lookup on language_name 
+| where NOT language_code <= 50 OR language_name == "English"
+| where language_code > 50 
+| where language_name == "English"
+;
+
+language_name:keyword | language_code:integer 
+;
+
+
+// https://github.com/elastic/elasticsearch/issues/136599
+lookupJoinWithSemanticFilterDeduplicationComplex
+required_capability: join_lookup_v12
+required_capability: lookup_join_semantic_filter_dedup
+
+from languages_lookup_non_unique_ke* 
+| enrich languages_policy on language_name 
+| sort country DESC, language_code ASC NULLS LAST, language_name NULLS FIRST 
+| limit 14081 
+| keep `language_name`, `country`, *, `country.keyword` 
+| mv_expand country.keyword 
+| grok language_name "%{WORD:YEcxuCOHsGzb} %{WORD:country.keyword}"
+| rename country.keyword as other1
+| rename language_code as other2 
+| lookup join multi_column_joinable_lookup on other1, other2 
+| where  NOT id_int <= 50 OR  NOT false AND  NOT other2 == 50 
+| where false AND  NOT true OR id_int > 50 AND true 
+| where  NOT false AND other2 == 50
+;
+
+warning:Line 13:24: evaluation of [other2 == 50] failed, treating result as null. Only first 20 failures recorded.
+warning:Line 13:24: java.lang.IllegalArgumentException: single-value function encountered multi-value
+
+language_name:keyword | country:text | other2:integer | YEcxuCOHsGzb:keyword | other1:keyword |id_int:integer | ip_addr:ip |is_active_bool:boolean | name_str:keyword
+
+;

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

@@ -1629,6 +1629,14 @@ public enum Cap {
          * Temporarily forbid the use of an explicit or implicit LIMIT before INLINE STATS.
          */
         FORBID_LIMIT_BEFORE_INLINE_STATS(INLINE_STATS.enabled),
+
+        /**
+         * Fix for lookup join filter pushdown not using semantic equality.
+         * This prevents duplicate filters from being pushed down when they are semantically equivalent, causing an infinite loop where
+         * BooleanSimplification will simplify the original and duplicate filters, so they'll be pushed down again...
+         */
+        LOOKUP_JOIN_SEMANTIC_FILTER_DEDUP,
+
         // Last capability should still have a comma for fewer merge conflicts when adding new ones :)
         // This comment prevents the semicolon from being on the previous capability when Spotless formats the file.
         ;

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

@@ -171,7 +171,9 @@ private static LogicalPlan pushDownPastJoin(Filter filter, Join join, FoldContex
 
                         List<Expression> existingFilters = new ArrayList<>(Predicates.splitAnd(existingRightFilter.condition()));
                         int sizeBefore = existingFilters.size();
-                        rightPushableFilters.stream().filter(e -> existingFilters.contains(e) == false).forEach(existingFilters::add);
+                        rightPushableFilters.stream()
+                            .filter(e -> existingFilters.stream().anyMatch(x -> x.semanticEquals(e)) == false)
+                            .forEach(existingFilters::add);
                         if (sizeBefore != existingFilters.size()) {
                             right = existingRightFilter.with(Predicates.combineAnd(existingFilters));
                             join = (Join) join.replaceRight(right);

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

@@ -89,6 +89,7 @@
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.Equals;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.EsqlBinaryComparison;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.GreaterThan;
+import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.GreaterThanOrEqual;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.In;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.InsensitiveEquals;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.LessThan;
@@ -9233,4 +9234,57 @@ public void testLookupJoinExpressionSameAttrsDifferentConditions() {
         // Verify the right side of join is EsRelation with LOOKUP
         as(join.right(), EsRelation.class);
     }
+
+    /**
+     * Project[[_meta_field{f}#16, emp_no{f}#10, first_name{f}#11 AS language_name#4, gender{f}#12, hire_date{f}#17, job{f}#1
+     * 8, job.raw{f}#19, languages{f}#13, last_name{f}#14, long_noidx{f}#20, salary{f}#15, language_code{f}#21]]
+     * \_Limit[1000[INTEGER],false]
+     *   \_Filter[NOT(language_code{f}#21 >= 50[INTEGER])]
+     *     \_Join[LEFT,[first_name{f}#11],[language_name{f}#22],null]
+     *       |_Filter[first_name{f}#11 == [KEYWORD]]
+     *       | \_EsRelation[test][_meta_field{f}#16, emp_no{f}#10, first_name{f}#11, ..]
+     *       \_Filter[language_code{f}#21 < 50[INTEGER]]
+     *         \_EsRelation[languages_lookup][LOOKUP][language_code{f}#21, language_name{f}#22]
+     */
+    public void LookupJoinSemanticFilterDeupPushdown() {
+        LogicalPlan plan = optimizedPlan("""
+            from test
+            | rename first_name as language_name
+            | lookup join languages_lookup on language_name
+            | where NOT language_code >= 50 OR language_name == ""
+            | where language_code < 50
+            | where language_name == ""
+            """);
+
+        var project = as(plan, Project.class);
+        var limit = as(project.child(), Limit.class);
+        var filter = as(limit.child(), Filter.class);
+
+        // Verify the top-level filter is NOT(language_code >= 50)
+        var not = as(filter.condition(), Not.class);
+        var gte = as(not.field(), GreaterThanOrEqual.class);
+        assertThat(Expressions.name(gte.left()), equalTo("language_code"));
+        assertThat(gte.right().fold(FoldContext.small()), equalTo(50));
+
+        // Verify the join structure
+        var join = as(filter.child(), Join.class);
+        assertThat(join.config().type(), equalTo(JoinTypes.LEFT));
+
+        // Verify left side has filter for language_name == ""
+        var leftFilter = as(join.left(), Filter.class);
+        var leftEquals = as(leftFilter.condition(), Equals.class);
+        assertThat(Expressions.name(leftEquals.left()), equalTo("first_name"));
+        assertThat(leftEquals.right().fold(FoldContext.small()), equalTo(new BytesRef("")));
+
+        var leftRelation = as(leftFilter.child(), EsRelation.class);
+
+        // Verify right side has filter for language_code < 50
+        var rightFilter = as(join.right(), Filter.class);
+        var rightLt = as(rightFilter.condition(), LessThan.class);
+        assertThat(Expressions.name(rightLt.left()), equalTo("language_code"));
+        assertThat(rightLt.right().fold(FoldContext.small()), equalTo(50));
+
+        var rightRelation = as(rightFilter.child(), EsRelation.class);
+    }
+
 }