Switch to storing the filter in Join

Author: julian-elastic

docs/changelog/132889.yaml

@@ -0,0 +1,6 @@
+pr: 132889
+summary: Improve Expanding Lookup Join performance by pushing a filter to the lookup
+  join
+area: "ES|QL, Performance"
+type: enhancement
+issues: []

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/join/InlineJoin.java

@@ -7,12 +7,14 @@
 
 package org.elasticsearch.xpack.esql.plan.logical.join;
 
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockUtils;
 import org.elasticsearch.xpack.esql.core.expression.Alias;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
 import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
 import org.elasticsearch.xpack.esql.core.tree.Source;
@@ -152,7 +154,7 @@ public InlineJoin(
         List<Attribute> leftFields,
         List<Attribute> rightFields
     ) {
-        super(source, left, right, type, matchFields, leftFields, rightFields);
+        super(source, left, right, type, matchFields, leftFields, rightFields, false, null);
     }
 
     private static InlineJoin readFrom(StreamInput in) throws IOException {
@@ -161,6 +163,9 @@ private static InlineJoin readFrom(StreamInput in) throws IOException {
         LogicalPlan left = in.readNamedWriteable(LogicalPlan.class);
         LogicalPlan right = in.readNamedWriteable(LogicalPlan.class);
         JoinConfig config = new JoinConfig(in);
+        if (in.getTransportVersion().onOrAfter(TransportVersions.ESQL_LOOKUP_JOIN_PRE_JOIN_FILTER)) {
+            Expression ignored = in.readOptionalNamedWriteable(Expression.class);
+        }
         return new InlineJoin(source, left, replaceStub(left, right), config);
     }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/join/Join.java

@@ -120,10 +120,14 @@ public Join(
         JoinType type,
         List<Attribute> matchFields,
         List<Attribute> leftFields,
-        List<Attribute> rightFields
+        List<Attribute> rightFields,
+        boolean isRemote,
+        Expression optionalRightHandFilters
     ) {
         super(source, left, right);
         this.config = new JoinConfig(type, matchFields, leftFields, rightFields);
+        this.isRemote = isRemote;
+        this.optionalRightHandFilters = optionalRightHandFilters;
     }
 
     public Join(StreamInput in) throws IOException {
@@ -170,7 +174,9 @@ protected NodeInfo<Join> info() {
             config.type(),
             config.matchFields(),
             config.leftFields(),
-            config.rightFields()
+            config.rightFields(),
+            isRemote(),
+            optionalRightHandFilters()
         );
     }
 

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

@@ -1923,10 +1923,9 @@ public void testDoNotPushDownIsNullFilterPastLookupJoin() {
      * uages{f}#10 AS language_code#4, last_name{f}#11, long_noidx{f}#17, salary{f}#12, language_name{f}#19]]
      * \_Limit[1000[INTEGER],false]
      *   \_Filter[language_name{f}#19 > a[KEYWORD]]
-     *     \_Join[LEFT,[languages{f}#10],[languages{f}#10],[language_code{f}#18]]
+     *     \_Join[LEFT,[languages{f}#10],[languages{f}#10],[language_code{f}#18],false,language_name{f}#19 > a[KEYWORD]]
      *       |_EsRelation[test][_meta_field{f}#13, emp_no{f}#7, first_name{f}#8, ge..]
-     *       \_Filter[language_name{f}#19 > a[KEYWORD]]
-     *         \_EsRelation[languages_lookup][LOOKUP][language_code{f}#18, language_name{f}#19]
+     *       \_EsRelation[languages_lookup][LOOKUP][language_code{f}#18, language_name{f}#19]
      */
     public void testPushDownGreaterThanFilterPastLookupJoin() {
         var plan = plan("""
@@ -1940,8 +1939,8 @@ public void testPushDownGreaterThanFilterPastLookupJoin() {
         var limit = as(project.child(), Limit.class);
         var filter = as(limit.child(), Filter.class);
         var join = as(filter.child(), Join.class);
-        var right = as(join.right(), Filter.class);
-        assertThat(right.condition().toString(), is("language_name > \"a\""));
+        var right = as(join.right(), EsRelation.class);
+        assertThat(join.optionalRightHandFilters().toString(), is("language_name > \"a\""));
     }
 
     /**
@@ -1970,13 +1969,12 @@ public void testDoNotPushDownCoalesceFilterPastLookupJoin() {
 
     /**
      * Project[[_meta_field{f}#13, emp_no{f}#7, first_name{f}#8, gender{f}#9, hire_date{f}#14, job{f}#15, job.raw{f}#16, lang
-     * uages{f}#10 AS language_code#4, last_name{f}#11, long_noidx{f}#17, salary{f}#12, language_name{f}#19]]
+     *uages{f}#10 AS language_code#4, last_name{f}#11, long_noidx{f}#17, salary{f}#12, language_name{f}#19]]
      * \_Limit[1000[INTEGER],false]
      *   \_Filter[ISNOTNULL(language_name{f}#19)]
-     *     \_Join[LEFT,[languages{f}#10],[languages{f}#10],[language_code{f}#18]]
+     *     \_Join[LEFT,[languages{f}#10],[languages{f}#10],[language_code{f}#18],false,ISNOTNULL(language_name{f}#19)]
      *       |_EsRelation[test][_meta_field{f}#13, emp_no{f}#7, first_name{f}#8, ge..]
-     *       \_Filter[ISNOTNULL(language_name{f}#19)]
-     *         \_EsRelation[languages_lookup][LOOKUP][language_code{f}#18, language_name{f}#19]
+     *       \_EsRelation[languages_lookup][LOOKUP][language_code{f}#18, language_name{f}#19]
      */
     public void testPushDownIsNotNullFilterPastLookupJoin() {
         var plan = plan("""
@@ -1990,8 +1988,8 @@ public void testPushDownIsNotNullFilterPastLookupJoin() {
         var limit = as(project.child(), Limit.class);
         var filter = as(limit.child(), Filter.class);
         var join = as(filter.child(), Join.class);
-        var right = as(join.right(), Filter.class);
-        assertThat(right.condition().toString(), is("language_name IS NOT NULL"));
+        var right = as(join.right(), EsRelation.class);
+        assertThat(join.optionalRightHandFilters().toString(), is("language_name IS NOT NULL"));
     }
 
     /**
@@ -7150,14 +7148,13 @@ public void testLookupJoinPushDownFilterOnLeftSideField() {
      * Expects
      *
      * <pre>{@code
-     * Project[[_meta_field{f}#13, emp_no{f}#7, first_name{f}#8, gender{f}#9, hire_date{f}#14, job{f}#15, job.raw{f}#16,
-     *          languages{f}#10 AS language_code#4, last_name{f}#11, long_noidx{f}#17, salary{f}#12, language_name{f}#19]]
+     * Project[[_meta_field{f}#13, emp_no{f}#7, first_name{f}#8, gender{f}#9, hire_date{f}#14, job{f}#15, job.raw{f}#16, lang
+     * uages{f}#10 AS language_code#4, last_name{f}#11, long_noidx{f}#17, salary{f}#12, language_name{f}#19]]
      * \_Limit[1000[INTEGER],false]
-     *   \_Filter[language_name{f}#19 == [45 6e 67 6c 69 73 68][KEYWORD]]
-     *     \_Join[LEFT,[languages{f}#10],[languages{f}#10],[language_code{f}#18]]
+     *   \_Filter[language_name{f}#19 == English[KEYWORD]]
+     *     \_Join[LEFT,[languages{f}#10],[languages{f}#10],[language_code{f}#18],false,language_name{f}#19 == English[KEYWORD]]
      *       |_EsRelation[test][_meta_field{f}#13, emp_no{f}#7, first_name{f}#8, ge..]
-     *       \_Filter[language_name{f}#19 == English[KEYWORD]]
-     *         \_EsRelation[languages_lookup][LOOKUP][language_code{f}#18, language_name{f}#19]
+     *       \_EsRelation[languages_lookup][LOOKUP][language_code{f}#18, language_name{f}#19]
      * }</pre>
      */
     public void testLookupJoinPushDownDisabledForLookupField() {
@@ -7186,9 +7183,8 @@ public void testLookupJoinPushDownDisabledForLookupField() {
         assertThat(join.config().type(), equalTo(JoinTypes.LEFT));
 
         var leftRel = as(join.left(), EsRelation.class);
-        var filterRight = as(join.right(), Filter.class);
-        assertEquals("language_name == \"English\"", filterRight.condition().toString());
-        var joinRightEsRelation = as(filterRight.child(), EsRelation.class);
+        assertEquals("language_name == \"English\"", join.optionalRightHandFilters().toString());
+        var joinRightEsRelation = as(join.right(), EsRelation.class);
 
     }
 
@@ -7198,11 +7194,11 @@ public void testLookupJoinPushDownDisabledForLookupField() {
      * Expects
      *
      * <pre>{@code
-     * Project[[_meta_field{f}#14, emp_no{f}#8, first_name{f}#9, gender{f}#10, hire_date{f}#15, job{f}#16, job.raw{f}#17,
-     *          languages{f}#11 AS language_code#4, last_name{f}#12, long_noidx{f}#18, salary{f}#13, language_name{f}#20]]
+     * Project[[_meta_field{f}#14, emp_no{f}#8, first_name{f}#9, gender{f}#10, hire_date{f}#15, job{f}#16, job.raw{f}#17, lan
+     * guages{f}#11 AS language_code#4, last_name{f}#12, long_noidx{f}#18, salary{f}#13, language_name{f}#20]]
      * \_Limit[1000[INTEGER],false]
-     *   \_Filter[language_name{f}#20 == [45 6e 67 6c 69 73 68][KEYWORD]]
-     *     \_Join[LEFT,[languages{f}#11],[languages{f}#11],[language_code{f}#19]]
+     *   \_Filter[language_name{f}#20 == English[KEYWORD]]
+     *     \_Join[LEFT,[languages{f}#11],[languages{f}#11],[language_code{f}#19],false,language_name{f}#20 == English[KEYWORD]]
      *       |_Filter[emp_no{f}#8 > 1[INTEGER]]
      *       | \_EsRelation[test][_meta_field{f}#14, emp_no{f}#8, first_name{f}#9, ge..]
      *       \_EsRelation[languages_lookup][LOOKUP][language_code{f}#19, language_name{f}#20]
@@ -7242,9 +7238,8 @@ public void testLookupJoinPushDownSeparatedForConjunctionBetweenLeftAndRightFiel
         assertThat(literal.value(), equalTo(1));
 
         var leftRel = as(filter.child(), EsRelation.class);
-        var filterRight = as(join.right(), Filter.class);
-        assertEquals("language_name == \"English\"", filterRight.condition().toString());
-        var rightRel = as(filterRight.child(), EsRelation.class);
+        assertEquals("language_name == \"English\"", join.optionalRightHandFilters().toString());
+        var rightRel = as(join.right(), EsRelation.class);
     }
 
     /**