diff --git a/docs/changelog/136104.yaml b/docs/changelog/136104.yaml new file mode 100644 index 0000000000000..7ec37704f51be --- /dev/null +++ b/docs/changelog/136104.yaml @@ -0,0 +1,5 @@ +pr: 136104 +summary: Add support for Full Text Functions for Lookup Join +area: ES|QL +type: enhancement +issues: [] diff --git a/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java b/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java index d5a12db64d291..50ada817970a6 100644 --- a/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java +++ b/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java @@ -861,11 +861,22 @@ private Map lookupExplosion( } } if (lookupEntries != lookupEntriesToKeep) { - // add a filter to reduce the number of matches - // we add both a Lucene pushable filter and a non-pushable filter - // this is to make sure that even if there are non-pushable filters the pushable filters is still applied - query.append(" | WHERE ABS(filter_key) > -1 AND filter_key < ").append(lookupEntriesToKeep); - + boolean applyAsExpressionJoinFilter = expressionBasedJoin && randomBoolean(); + // we randomly add the filter after the join or as part of the join + // in both cases we should have the same amount of results + if (applyAsExpressionJoinFilter == false) { + // add a filter after the join to reduce the number of matches + // we add both a Lucene pushable filter and a non-pushable filter + // this is to make sure that even if there are non-pushable filters the pushable filters is still applied + query.append(" | WHERE ABS(filter_key) > -1 AND filter_key < ").append(lookupEntriesToKeep); + } else { + // apply the filter as part of the join + // then we filter out the rows that do not match the filter after + // so the number of rows is the same as in the field based join case + // and can get the same number of rows for verification purposes + query.append(" AND filter_key < ").append(lookupEntriesToKeep); + query.append(" | WHERE filter_key IS NOT NULL "); + } } query.append(" | STATS COUNT(location) | LIMIT 100\"}"); return responseAsMap(query(query.toString(), null)); diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/lookup-join-expression.csv-spec b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/lookup-join-expression.csv-spec index 44a16bbcf0c31..2cf80d806840c 100644 --- a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/lookup-join-expression.csv-spec +++ b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/lookup-join-expression.csv-spec @@ -725,6 +725,9 @@ id_int:integer | name_str:keyword | extra1:keyword | other1:keyword | other2:int 14 | Nina | foo2 | omicron | 15000 ; + + + lookupJoinExpressionOnUnionTypes required_capability: join_lookup_v12 required_capability: lookup_join_on_boolean_expression @@ -747,3 +750,303 @@ apps | 2 | French apps_short | 1 | English apps_short | 2 | French ; + +lookupJoinWithGreaterThanCondition +required_capability: join_lookup_v12 +required_capability: lookup_join_on_boolean_expression_v2 + +FROM multi_column_joinable +| RENAME id_int AS id_left, is_active_bool AS is_active_left +| LOOKUP JOIN multi_column_joinable_lookup ON id_int == id_left and is_active_left == is_active_bool AND other2 > 10000 +| KEEP id_left, name_str, extra1, other1, other2 +| SORT id_left, name_str, extra1, other1, other2 +; + +warning:Line 3:3: evaluation of [LOOKUP JOIN multi_column_joinable_lookup ON id_int == id_left and is_active_left == is_active_bool AND other2 > 10000] failed, treating result as null. Only first 20 failures recorded. +warning:Line 3:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +id_left:integer | name_str:keyword | extra1:keyword | other1:keyword | other2:integer +1 | null | foo | null | null +[1, 19, 21] | null | zyx | null | null +2 | null | bar | null | null +3 | null | baz | null | null +4 | null | qux | null | null +5 | null | quux | null | null +6 | null | corge | null | null +7 | null | grault | null | null +8 | Hank | garply | lambda | 11000 +9 | null | waldo | null | null +10 | null | fred | null | null +12 | Liam | xyzzy | nu | 13000 +13 | Mia | thud | xi | 14000 +14 | Nina | foo2 | omicron | 15000 +15 | null | bar2 | null | null +[17, 18] | null | xyz | null | null +null | null | plugh | null | null +; + +lookupJoinWithLikeCondition +required_capability: join_lookup_v12 +required_capability: lookup_join_on_boolean_expression_v2 + +FROM multi_column_joinable +| RENAME id_int AS id_left, is_active_bool AS is_active_left +| LOOKUP JOIN multi_column_joinable_lookup ON id_int == id_left and is_active_left == is_active_bool AND other1 like "*ta" +| KEEP id_left, name_str, extra1, other1, other2 +| SORT id_left, name_str, extra1, other1, other2 +; + +warning:Line 3:3: evaluation of [LOOKUP JOIN multi_column_joinable_lookup ON id_int == id_left and is_active_left == is_active_bool AND other1 like \"*ta\"] failed, treating result as null. Only first 20 failures recorded. +warning:Line 3:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +id_left:integer | name_str:keyword | extra1:keyword | other1:keyword | other2:integer +1 | Alice | foo | beta | 2000 +[1, 19, 21] | null | zyx | null | null +2 | null | bar | null | null +3 | Charlie | baz | delta | 4000 +4 | David | qux | zeta | 6000 +5 | Eve | quux | eta | 7000 +5 | Eve | quux | theta | 8000 +6 | null | corge | iota | 9000 +7 | null | grault | null | null +8 | null | garply | null | null +9 | null | waldo | null | null +10 | null | fred | null | null +12 | null | xyzzy | null | null +13 | null | thud | null | null +14 | null | foo2 | null | null +15 | null | bar2 | null | null +[17, 18] | null | xyz | null | null +null | null | plugh | null | null +; + +lookupJoinWithOrOfLikeGt +required_capability: join_lookup_v12 +required_capability: lookup_join_on_boolean_expression_v2 + +FROM multi_column_joinable +| RENAME id_int AS id_left, is_active_bool AS is_active_left +| LOOKUP JOIN multi_column_joinable_lookup ON id_int == id_left and is_active_left == is_active_bool AND (other1 like "*ta" OR other2 > 10000) +| KEEP id_left, name_str, extra1, other1, other2 +| SORT id_left, name_str, extra1, other1, other2 +; + +warning:Line 3:3: evaluation of [LOOKUP JOIN multi_column_joinable_lookup ON id_int == id_left and is_active_left == is_active_bool AND (other1 like \"*ta\" OR other2 > 10000)] failed, treating result as null. Only first 20 failures recorded. +warning:Line 3:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +id_left:integer | name_str:keyword | extra1:keyword | other1:keyword | other2:integer +1 | Alice | foo | beta | 2000 +[1, 19, 21] | null | zyx | null | null +2 | null | bar | null | null +3 | Charlie | baz | delta | 4000 +4 | David | qux | zeta | 6000 +5 | Eve | quux | eta | 7000 +5 | Eve | quux | theta | 8000 +6 | null | corge | iota | 9000 +7 | null | grault | null | null +8 | Hank | garply | lambda | 11000 +9 | null | waldo | null | null +10 | null | fred | null | null +12 | Liam | xyzzy | nu | 13000 +13 | Mia | thud | xi | 14000 +14 | Nina | foo2 | omicron | 15000 +15 | null | bar2 | null | null +[17, 18] | null | xyz | null | null +null | null | plugh | null | null +; + +lookupJoinExpressionWithMatch +required_capability: join_lookup_v12 +required_capability: lookup_join_on_boolean_expression_v2 +FROM multi_column_joinable +| RENAME id_int AS id_left, is_active_bool AS is_active_left +| LOOKUP JOIN multi_column_joinable_lookup ON MATCH(other1, "beta") AND id_int == id_left and is_active_left == is_active_bool +| WHERE other2 IS NOT NULL +| KEEP id_left, name_str, extra1, other1, other2 +| SORT id_left, name_str, extra1, other1, other2 +; + +warning:Line 3:3: evaluation of [LOOKUP JOIN multi_column_joinable_lookup ON MATCH(other1, \"beta\") AND id_int == id_left and is_active_left == is_active_bool] failed, treating result as null. Only first 20 failures recorded. +warning:Line 3:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +id_left:integer | name_str:keyword | extra1:keyword | other1:keyword | other2:integer +1 | Alice | foo | beta | 2000 +; + +lookupJoinOnSameFieldTwiceWithOrNot +required_capability: join_lookup_v12 +required_capability: lookup_join_on_boolean_expression_v2 + +FROM multi_column_joinable +| RENAME id_int AS id_left, name_str AS name_left, is_active_bool AS is_active_left, ip_addr AS ip_addr_left +| LOOKUP JOIN multi_column_joinable_lookup ON (other2 < 12000 OR NOT (other1 != "omicron" AND other1 != "nu")) AND id_left == id_int AND name_left == name_str AND id_left < other2 +| KEEP id_left, name_left, extra1, other1, other2 +| SORT id_left, name_left, extra1, other1, other2 +; + +warning:Line 3:3: evaluation of [LOOKUP JOIN multi_column_joinable_lookup ON (other2 < 12000 OR NOT (other1 != \"omicron\" AND other1 != \"nu\")) AND id_left == id_int AND name_left == name_str AND id_left < other2] failed, treating result as null. Only first 20 failures recorded. +warning:Line 3:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +id_left:integer | name_left:keyword | extra1:keyword | other1:keyword | other2:integer +1 | Alice | foo | alpha | 1000 +1 | Alice | foo | beta | 2000 +[1, 19, 21] | Sophia | zyx | null | null +2 | Bob | bar | gamma | 3000 +3 | Charlie | baz | delta | 4000 +3 | Charlie | baz | epsilon | 5000 +4 | David | qux | zeta | 6000 +5 | Eve | quux | eta | 7000 +5 | Eve | quux | theta | 8000 +6 | null | corge | null | null +7 | Grace | grault | kappa | 10000 +8 | Hank | garply | lambda | 11000 +9 | Ivy | waldo | null | null +10 | John | fred | null | null +12 | Liam | xyzzy | nu | 13000 +13 | Mia | thud | null | null +14 | Nina | foo2 | omicron | 15000 +15 | Oscar | bar2 | null | null +[17, 18] | Olivia | xyz | null | null +null | Kate | plugh | null | null +; + + +lookupJoinOnSameFieldWithPushableRightFilterAfter +required_capability: join_lookup_v12 +required_capability: lookup_join_on_boolean_expression_v2 + +FROM multi_column_joinable +| RENAME id_int AS id_left, name_str AS name_left, is_active_bool AS is_active_left, ip_addr AS ip_addr_left +| LOOKUP JOIN multi_column_joinable_lookup ON (other2 < 12000 OR NOT (other1 != "omicron" AND other1 != "nu")) AND id_left == id_int AND name_left == name_str AND id_left < other2 +| WHERE other1 like ("a*", "b*", "o*") +| KEEP id_left, name_left, extra1, other1, other2 +| SORT id_left, name_left, extra1, other1, other2 +; + +warning:Line 3:3: evaluation of [LOOKUP JOIN multi_column_joinable_lookup ON (other2 < 12000 OR NOT (other1 != \"omicron\" AND other1 != \"nu\")) AND id_left == id_int AND name_left == name_str AND id_left < other2] failed, treating result as null. Only first 20 failures recorded. +warning:Line 3:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +id_left:integer | name_left:keyword | extra1:keyword | other1:keyword | other2:integer +1 | Alice | foo | alpha | 1000 +1 | Alice | foo | beta | 2000 +14 | Nina | foo2 | omicron | 15000 +; + +twoLookupJoinsInSameQueryOtherFilter +required_capability: join_lookup_v12 +required_capability: lookup_join_on_boolean_expression_v2 + +FROM multi_column_joinable +| WHERE id_int == 1 +| RENAME id_int AS id_left, name_str AS name_left +| LOOKUP JOIN multi_column_joinable_lookup ON id_left == id_int AND name_left == name_str +| RENAME other1 AS other1_from_first_join, id_int AS id_from_first_join, name_str AS name_from_first_join +| LOOKUP JOIN multi_column_joinable_lookup ON id_left == id_int AND other1_from_first_join != other1 AND other1 like ("a*", "c*") +| KEEP id_left, name_left, other1_from_first_join, other1 +| SORT id_left, name_left, other1_from_first_join, other1 +; + +warning:Line 2:9: evaluation of [id_int == 1] failed, treating result as null. Only first 20 failures recorded. +warning:Line 2:9: java.lang.IllegalArgumentException: single-value function encountered multi-value + +id_left:integer | name_left:keyword | other1_from_first_join:keyword | other1:keyword +1 | Alice | alpha | null +1 | Alice | beta | alpha +; + +lookupJoinExpressionWithTerm +required_capability: join_lookup_v12 +required_capability: lookup_join_on_boolean_expression_v2 + +FROM multi_column_joinable +| RENAME id_int AS id_left, is_active_bool AS is_active_left +| LOOKUP JOIN multi_column_joinable_lookup ON TERM(other1, "beta") AND id_int == id_left and is_active_left == is_active_bool +| WHERE other2 IS NOT NULL +| KEEP id_left, name_str, extra1, other1, other2 +| SORT id_left, name_str, extra1, other1, other2 +; + +warning:Line 3:3: evaluation of [LOOKUP JOIN multi_column_joinable_lookup ON TERM(other1, \"beta\") AND id_int == id_left and is_active_left == is_active_bool] failed, treating result as null. Only first 20 failures recorded. +warning:Line 3:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +id_left:integer | name_str:keyword | extra1:keyword | other1:keyword | other2:integer +1 | Alice | foo | beta | 2000 +; + +lookupJoinExpressionWithQueryString +required_capability: join_lookup_v12 +required_capability: lookup_join_on_boolean_expression_v2 + +FROM multi_column_joinable +| RENAME id_int AS id_left, is_active_bool AS is_active_left +| LOOKUP JOIN multi_column_joinable_lookup ON QSTR("other1:alpha OR other1:beta") AND id_int == id_left and is_active_left == is_active_bool +| WHERE other2 IS NOT NULL +| KEEP id_left, name_str, extra1, other1, other2 +| SORT id_left, name_str, extra1, other1, other2 +; + +warning:Line 3:3: evaluation of [LOOKUP JOIN multi_column_joinable_lookup ON QSTR(\"other1:alpha OR other1:beta\") AND id_int == id_left and is_active_left == is_active_bool] failed, treating result as null. Only first 20 failures recorded. +warning:Line 3:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +id_left:integer | name_str:keyword | extra1:keyword | other1:keyword | other2:integer +1 | Alice | foo | alpha | 1000 +1 | Alice | foo | beta | 2000 +; + +lookupJoinExpressionWithKql +required_capability: join_lookup_v12 +required_capability: lookup_join_on_boolean_expression_v2 + +FROM multi_column_joinable +| RENAME id_int AS id_left, is_active_bool AS is_active_left +| LOOKUP JOIN multi_column_joinable_lookup ON KQL("other1:alpha OR other1:beta") AND id_int == id_left and is_active_left == is_active_bool +| WHERE other2 IS NOT NULL +| KEEP id_left, name_str, extra1, other1, other2 +| SORT id_left, name_str, extra1, other1, other2 +; + +warning:Line 3:3: evaluation of [LOOKUP JOIN multi_column_joinable_lookup ON KQL(\"other1:alpha OR other1:beta\") AND id_int == id_left and is_active_left == is_active_bool] failed, treating result as null. Only first 20 failures recorded. +warning:Line 3:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +id_left:integer | name_str:keyword | extra1:keyword | other1:keyword | other2:integer +1 | Alice | foo | alpha | 1000 +1 | Alice | foo | beta | 2000 +; + +lookupJoinExpressionWithMultiMatch +required_capability: join_lookup_v12 +required_capability: lookup_join_on_boolean_expression_v2 + +FROM multi_column_joinable +| RENAME id_int AS id_left, is_active_bool AS is_active_left, name_str AS name_str_left +| LOOKUP JOIN multi_column_joinable_lookup ON MULTI_MATCH("beta", other1, name_str) AND id_int == id_left and is_active_left == is_active_bool +| WHERE other2 IS NOT NULL +| KEEP id_left, name_str, extra1, other1, other2 +| SORT id_left, name_str, extra1, other1, other2 +; + +warning:Line 3:3: evaluation of [LOOKUP JOIN multi_column_joinable_lookup ON MULTI_MATCH(\"beta\", other1, name_str) AND id_int == id_left and is_active_left == is_active_bool] failed, treating result as null. Only first 20 failures recorded. +warning:Line 3:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +id_left:integer | name_str:keyword | extra1:keyword | other1:keyword | other2:integer +1 | Alice | foo | beta | 2000 +; + +lookupJoinExpressionWithMatchPhrase +required_capability: join_lookup_v12 +required_capability: lookup_join_on_boolean_expression_v2 + +FROM multi_column_joinable +| RENAME id_int AS id_left, is_active_bool AS is_active_left +| LOOKUP JOIN multi_column_joinable_lookup ON MATCH_PHRASE(other1, "beta") AND id_int == id_left and is_active_left == is_active_bool +| WHERE other2 IS NOT NULL +| KEEP id_left, name_str, extra1, other1, other2 +| SORT id_left, name_str, extra1, other1, other2 +; + +warning:Line 3:3: evaluation of [LOOKUP JOIN multi_column_joinable_lookup ON MATCH_PHRASE(other1, \"beta\") AND id_int == id_left and is_active_left == is_active_bool] failed, treating result as null. Only first 20 failures recorded. +warning:Line 3:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +id_left:integer | name_str:keyword | extra1:keyword | other1:keyword | other2:integer +1 | Alice | foo | beta | 2000 +; diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/LookupFromIndexIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/LookupFromIndexIT.java index a3c52374ef859..176945b366f59 100644 --- a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/LookupFromIndexIT.java +++ b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/LookupFromIndexIT.java @@ -250,7 +250,7 @@ private PhysicalPlan buildGreaterThanFilter(long value) { return new FragmentExec(filter); } - private void runLookup(List keyTypes, PopulateIndices populateIndices, PhysicalPlan filters) throws IOException { + private void runLookup(List keyTypes, PopulateIndices populateIndices, PhysicalPlan pushedDownFilter) throws IOException { String[] fieldMappers = new String[keyTypes.size() * 2]; for (int i = 0; i < keyTypes.size(); i++) { fieldMappers[2 * i] = "key" + i; @@ -283,17 +283,8 @@ private void runLookup(List keyTypes, PopulateIndices populateIndices, client().admin().cluster().prepareHealth(TEST_REQUEST_TIMEOUT).setWaitForGreenStatus().get(); Predicate filterPredicate = l -> true; - if (filters instanceof FragmentExec fragmentExec) { - if (fragmentExec.fragment() instanceof Filter filter - && filter.condition() instanceof GreaterThan gt - && gt.left() instanceof FieldAttribute fa - && fa.name().equals("l") - && gt.right() instanceof Literal lit) { - long value = ((Number) lit.value()).longValue(); - filterPredicate = l -> l > value; - } else { - fail("Unsupported filter type in test baseline generation: " + filters); - } + if (pushedDownFilter instanceof FragmentExec fragmentExec && fragmentExec.fragment() instanceof Filter filter) { + filterPredicate = getPredicateFromFilter(filter); } int docCount = between(10, 1000); @@ -396,6 +387,16 @@ private void runLookup(List keyTypes, PopulateIndices populateIndices, new EsField("rkey" + i, keyTypes.get(i), Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE) ); joinOnConditions.add(new Equals(Source.EMPTY, leftAttr, rightAttr)); + // randomly decide to apply the filter as additional join on filter instead of pushed down filter + boolean applyAsJoinOnCondition = EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION_V2.isEnabled() + ? randomBoolean() + : false; + if (applyAsJoinOnCondition + && pushedDownFilter instanceof FragmentExec fragmentExec + && fragmentExec.fragment() instanceof Filter filter) { + joinOnConditions.add(filter.condition()); + pushedDownFilter = null; + } } } // the matchFields are shared for both types of join @@ -412,7 +413,7 @@ private void runLookup(List keyTypes, PopulateIndices populateIndices, "lookup", List.of(new Alias(Source.EMPTY, "l", new ReferenceAttribute(Source.EMPTY, "l", DataType.LONG))), Source.EMPTY, - filters, + pushedDownFilter, Predicates.combineAnd(joinOnConditions) ); DriverContext driverContext = driverContext(); @@ -478,6 +479,19 @@ protected void start(Driver driver, ActionListener driverListener) { } } + private static Predicate getPredicateFromFilter(Filter filter) { + if (filter.condition() instanceof GreaterThan gt + && gt.left() instanceof FieldAttribute fa + && fa.name().equals("l") + && gt.right() instanceof Literal lit) { + long value = ((Number) lit.value()).longValue(); + return l -> l > value; + } else { + fail("Unsupported filter type in test baseline generation: " + filter); + } + return null; + } + /** * Creates a {@link BigArrays} that tracks releases but doesn't throw circuit breaking exceptions. */ diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/plugin/MatchFunctionIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/plugin/MatchFunctionIT.java index 82124c4c85bb8..98a1a86df645b 100644 --- a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/plugin/MatchFunctionIT.java +++ b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/plugin/MatchFunctionIT.java @@ -313,6 +313,27 @@ public void testMatchWithLookupJoin() { ); } + public void testMatchWithLookupJoinOnMatch() { + var query = """ + FROM test + | rename id as id_left + | LOOKUP JOIN test_lookup ON id_left == id and MATCH(lookup_content, "fox") + | WHERE id > 0 + """; + try (var resp = run(query)) { + assertColumnNames(resp.columns(), List.of("content", "id_left", "id", "lookup_content")); + assertColumnTypes(resp.columns(), List.of("text", "integer", "integer", "text")); + // Should return rows where lookup_content matches "fox" (ids 1 and 6) + assertValues( + resp.values(), + List.of( + List.of("This is a brown fox", 1, 1, "This is a brown fox"), + List.of("The quick brown fox jumps over the lazy dog", 6, 6, "The quick brown fox jumps over the lazy dog") + ) + ); + } + } + static void createAndPopulateIndex(Consumer ensureYellow) { var indexName = "test"; var client = client().admin().indices(); @@ -341,5 +362,19 @@ static void createAndPopulateLookupIndex(IndicesAdminClient client, String looku .setSettings(Settings.builder().put("index.number_of_shards", 1).put("index.mode", "lookup")) .setMapping("id", "type=integer", "lookup_content", "type=text"); assertAcked(createRequest); + + // Populate the lookup index with test data + client().prepareBulk() + .add(new IndexRequest(lookupIndexName).id("1").source("id", 1, "lookup_content", "This is a brown fox")) + .add(new IndexRequest(lookupIndexName).id("2").source("id", 2, "lookup_content", "This is a brown dog")) + .add(new IndexRequest(lookupIndexName).id("3").source("id", 3, "lookup_content", "This dog is really brown")) + .add( + new IndexRequest(lookupIndexName).id("4") + .source("id", 4, "lookup_content", "The dog is brown but this document is very very long") + ) + .add(new IndexRequest(lookupIndexName).id("5").source("id", 5, "lookup_content", "There is also a white cat")) + .add(new IndexRequest(lookupIndexName).id("6").source("id", 6, "lookup_content", "The quick brown fox jumps over the lazy dog")) + .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) + .get(); } } diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java index cffbd17244d8c..1fed680ae1755 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java @@ -1451,6 +1451,11 @@ public enum Cap { INLINE_STATS_SUPPORTS_REMOTE(INLINESTATS_V11.enabled), INLINE_STATS_WITH_UNION_TYPES_IN_STUB_RELATION(INLINE_STATS.enabled), + /** + * Lookup join with Full Text Function or other Lucene Pushable condition + * to be applied to the lookup index used + */ + LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION_V2, /** * Support TS command in non-snapshot builds diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java index 58d74d0907c48..c0b846140b3b3 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java @@ -21,6 +21,7 @@ import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException; import org.elasticsearch.xpack.esql.VerificationException; import org.elasticsearch.xpack.esql.analysis.AnalyzerRules.ParameterizedAnalyzerRule; +import org.elasticsearch.xpack.esql.capabilities.TranslationAware; import org.elasticsearch.xpack.esql.common.Failure; import org.elasticsearch.xpack.esql.core.capabilities.Resolvables; import org.elasticsearch.xpack.esql.core.expression.Alias; @@ -102,6 +103,7 @@ import org.elasticsearch.xpack.esql.index.IndexResolution; import org.elasticsearch.xpack.esql.inference.ResolvedInference; import org.elasticsearch.xpack.esql.optimizer.rules.logical.SubstituteSurrogateExpressions; +import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.LucenePushdownPredicates; import org.elasticsearch.xpack.esql.parser.ParsingException; import org.elasticsearch.xpack.esql.plan.IndexPattern; import org.elasticsearch.xpack.esql.plan.logical.Aggregate; @@ -162,6 +164,7 @@ import static java.util.Collections.emptyList; import static java.util.Collections.singletonList; import static org.elasticsearch.xpack.core.enrich.EnrichPolicy.GEO_MATCH_TYPE; +import static org.elasticsearch.xpack.esql.capabilities.TranslationAware.translatable; import static org.elasticsearch.xpack.esql.core.type.DataType.AGGREGATE_METRIC_DOUBLE; import static org.elasticsearch.xpack.esql.core.type.DataType.BOOLEAN; import static org.elasticsearch.xpack.esql.core.type.DataType.DATETIME; @@ -724,50 +727,78 @@ private LogicalPlan resolveLookup(Lookup l, List childrenOutput) { private List resolveJoinFiltersAndSwapIfNeeded( List filters, - AttributeSet leftOutput, - AttributeSet rightOutput + AttributeSet leftChildOutput, + AttributeSet rightChildOutput, + List leftJoinKeysToPopulate, + List rightJoinKeysToPopulate ) { if (filters.isEmpty()) { return emptyList(); } - List childrenOutput = new ArrayList<>(leftOutput); - childrenOutput.addAll(rightOutput); + List childrenOutput = new ArrayList<>(leftChildOutput); + childrenOutput.addAll(rightChildOutput); List resolvedFilters = new ArrayList<>(filters.size()); for (Expression filter : filters) { Expression filterResolved = filter.transformUp(UnresolvedAttribute.class, ua -> maybeResolveAttribute(ua, childrenOutput)); - resolvedFilters.add(resolveAndOrientJoinCondition(filterResolved, leftOutput, rightOutput)); + Expression result = resolveAndOrientJoinCondition( + filterResolved, + leftChildOutput, + rightChildOutput, + leftJoinKeysToPopulate, + rightJoinKeysToPopulate + ); + resolvedFilters.add(result); } return resolvedFilters; } - private Expression resolveAndOrientJoinCondition(Expression condition, AttributeSet leftOutput, AttributeSet rightOutput) { + private Expression resolveAndOrientJoinCondition( + Expression condition, + AttributeSet leftChildOutput, + AttributeSet rightChildOutput, + List leftJoinKeysToPopulate, + List rightJoinKeysToPopulate + ) { if (condition instanceof EsqlBinaryComparison comp && comp.left() instanceof Attribute leftAttr && comp.right() instanceof Attribute rightAttr) { - boolean leftIsFromLeft = leftOutput.contains(leftAttr); - boolean rightIsFromRight = rightOutput.contains(rightAttr); + boolean leftIsFromLeft = leftChildOutput.contains(leftAttr); + boolean rightIsFromRight = rightChildOutput.contains(rightAttr); if (leftIsFromLeft && rightIsFromRight) { + leftJoinKeysToPopulate.add(leftAttr); + rightJoinKeysToPopulate.add(rightAttr); return comp; // Correct orientation } - boolean leftIsFromRight = rightOutput.contains(leftAttr); - boolean rightIsFromLeft = leftOutput.contains(rightAttr); + boolean leftIsFromRight = rightChildOutput.contains(leftAttr); + boolean rightIsFromLeft = leftChildOutput.contains(rightAttr); if (leftIsFromRight && rightIsFromLeft) { + leftJoinKeysToPopulate.add(rightAttr); + rightJoinKeysToPopulate.add(leftAttr); return comp.swapLeftAndRight(); // Swapped orientation } + } + return handleRightOnlyPushableFilter(condition, rightChildOutput); + } + + private Expression handleRightOnlyPushableFilter(Expression condition, AttributeSet rightChildOutput) { + if (isCompletelyRightSideAndTranslationAware(condition, rightChildOutput)) { + // The condition is completely on the right side and is translation aware, so it can be (potentially) pushed down + return condition; + } else { + // The condition cannot be used in the join on clause for now + // It is not a binary comparison between left and right attributes + // It is not using fields from the right side only and translation aware return new UnresolvedAttribute( condition.source(), "unsupported", - "Join condition must be between one attribute on the left side and " - + "one attribute on the right side of the join, but found: " - + condition.sourceText() + "Unsupported join filter expression:" + condition.sourceText() ); } - return condition; // Not a binary comparison between two attributes, no change needed. } private Join resolveLookupJoin(LookupJoin join) { @@ -787,37 +818,22 @@ private Join resolveLookupJoin(LookupJoin join) { List leftKeys = new ArrayList<>(); List rightKeys = new ArrayList<>(); List resolvedFilters = new ArrayList<>(); + Expression joinOnConditions = null; if (join.config().joinOnConditions() != null) { resolvedFilters = resolveJoinFiltersAndSwapIfNeeded( Predicates.splitAnd(join.config().joinOnConditions()), join.left().outputSet(), - join.right().outputSet() + join.right().outputSet(), + leftKeys, + rightKeys ); - // build leftKeys and rightKeys using the correct side of the resolvedFilters. - // resolveJoinFiltersAndSwapIfNeeded already put the left and right on the correct side - for (Expression expression : resolvedFilters) { - if (expression instanceof EsqlBinaryComparison binaryComparison - && binaryComparison.left() instanceof Attribute leftAttribute - && binaryComparison.right() instanceof Attribute rightAttribute) { - leftKeys.add(leftAttribute); - rightKeys.add(rightAttribute); - } else { - UnresolvedAttribute errorAttribute = new UnresolvedAttribute( - expression.source(), - "unsupported", - "Unsupported join filter expression:" + expression.sourceText() - ); - return join.withConfig(new JoinConfig(type, singletonList(errorAttribute), emptyList(), null)); - - } - } + joinOnConditions = Predicates.combineAndWithSource(resolvedFilters, join.config().joinOnConditions().source()); } else { // resolve the using columns against the left and the right side then assemble the new join config leftKeys = resolveUsingColumns(join.config().leftFields(), join.left().output(), "left"); rightKeys = resolveUsingColumns(join.config().rightFields(), join.right().output(), "right"); } - - config = new JoinConfig(type, leftKeys, rightKeys, Predicates.combineAnd(resolvedFilters)); + config = new JoinConfig(type, leftKeys, rightKeys, joinOnConditions); return new LookupJoin(join.source(), join.left(), join.right(), config, join.isRemote()); } else { // everything else is unsupported for now @@ -827,6 +843,33 @@ private Join resolveLookupJoin(LookupJoin join) { } } + private boolean isCompletelyRightSideAndTranslationAware(Expression expression, AttributeSet rightOutputSet) { + // Check if all references in the expression are from the right side + boolean isCompletelyRightSide = rightOutputSet.containsAll(expression.references()); + + if (isCompletelyRightSide == false) { + return false; + } + + // Check if the expression and all its subexpressions implement TranslationAware + // and are translatable to Lucene + return isTranslationAware(expression); + } + + private boolean isTranslationAware(Expression expression) { + // Check if the expression itself implements TranslationAware + if (expression instanceof TranslationAware == false) { + return false; + } + + // Check if the expression is translatable + TranslationAware.Translatable translatable = translatable(expression, LucenePushdownPredicates.DEFAULT); + if (translatable == TranslationAware.Translatable.NO) { + return false; + } + return true; + } + private LogicalPlan resolveFork(Fork fork, AnalyzerContext context) { // we align the outputs of the sub plans such that they have the same columns boolean changed = false; diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java index d881644849f2f..934dd94770e73 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java @@ -18,11 +18,13 @@ import org.elasticsearch.compute.operator.lookup.QueryList; import org.elasticsearch.index.mapper.MappedFieldType; import org.elasticsearch.index.query.QueryBuilder; +import org.elasticsearch.index.query.Rewriteable; import org.elasticsearch.index.query.SearchExecutionContext; import org.elasticsearch.search.internal.AliasFilter; import org.elasticsearch.xpack.esql.capabilities.TranslationAware; import org.elasticsearch.xpack.esql.core.expression.Attribute; import org.elasticsearch.xpack.esql.core.expression.Expression; +import org.elasticsearch.xpack.esql.core.type.DataType; import org.elasticsearch.xpack.esql.expression.predicate.Predicates; import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.Equals; import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.EsqlBinaryComparison; @@ -56,9 +58,10 @@ */ public class ExpressionQueryList implements LookupEnrichQueryGenerator { private final List queryLists; - private final List preJoinFilters = new ArrayList<>(); + private final List lucenePushableFilters = new ArrayList<>(); private final SearchExecutionContext context; private final AliasFilter aliasFilter; + private final LucenePushdownPredicates lucenePushdownPredicates; private ExpressionQueryList( List queryLists, @@ -70,6 +73,10 @@ private ExpressionQueryList( this.queryLists = new ArrayList<>(queryLists); this.context = context; this.aliasFilter = aliasFilter; + this.lucenePushdownPredicates = LucenePushdownPredicates.from( + SearchContextStats.from(List.of(context)), + new EsqlFlags(clusterService.getClusterSettings()) + ); buildPreJoinFilter(rightPreJoinPlan, clusterService); } @@ -141,98 +148,104 @@ private void buildJoinOnForExpressionJoin( ) { List expressions = Predicates.splitAnd(joinOnConditions); for (Expression expr : expressions) { - if (expr instanceof EsqlBinaryComparison binaryComparison) { - // the left side comes from the page that was sent to the lookup node - // the right side is the field from the lookup index - // check if the left side is in the matchFields - // if it is its corresponding page is the corresponding number in inputPage - Expression left = binaryComparison.left(); - if (left instanceof Attribute leftAttribute) { - boolean matched = false; - for (int i = 0; i < matchFields.size(); i++) { - if (matchFields.get(i).fieldName().equals(leftAttribute.name())) { - Block block = inputPage.getBlock(i); - Expression right = binaryComparison.right(); - if (right instanceof Attribute rightAttribute) { - MappedFieldType fieldType = context.getFieldType(rightAttribute.name()); - if (fieldType != null) { - // special handle Equals operator - // TermQuery is faster than BinaryComparisonQueryList, as it does less work per row - // so here we reuse the existing logic from field based join to build a termQueryList for Equals - if (binaryComparison instanceof Equals) { - QueryList termQueryForEquals = termQueryList( - fieldType, - context, - aliasFilter, - inputPage.getBlock(matchFields.get(i).channel()), - matchFields.get(i).type() - ).onlySingleValues(warnings, "LOOKUP JOIN encountered multi-value"); - queryLists.add(termQueryForEquals); - } else { - queryLists.add( - new BinaryComparisonQueryList( - fieldType, - context, - block, - binaryComparison, - clusterService, - aliasFilter, - warnings - ) - ); - } - matched = true; - break; - } else { - throw new IllegalStateException( - "Could not find field [" + rightAttribute.name() + "] in the lookup join index" - ); - } - } else { - throw new IllegalStateException( - "Only field from the right dataset are supported on the right of the join on condition but got: " + expr - ); - } - } - } - if (matched == false) { - throw new IllegalStateException( - "Could not find field [" + leftAttribute.name() + "] in the left side of the lookup join" - ); - } + boolean applied = applyAsLeftRightBinaryComparison(expr, matchFields, inputPage, clusterService, warnings); + if (applied == false) { + applied = applyAsRightSidePushableFilter(expr); + } + if (applied == false) { + throw new IllegalArgumentException("Cannot apply join condition: " + expr); + } + } + } + + private boolean applyAsRightSidePushableFilter(Expression filter) { + if (filter instanceof TranslationAware translationAware) { + if (TranslationAware.Translatable.YES.equals(translationAware.translatable(lucenePushdownPredicates))) { + QueryBuilder queryBuilder = translationAware.asQuery(lucenePushdownPredicates, TRANSLATOR_HANDLER).toQueryBuilder(); + // Rewrite the query builder to ensure doIndexMetadataRewrite is called + // Some functions, such as KQL require rewriting to work properly + try { + queryBuilder = Rewriteable.rewrite(queryBuilder, context, true); + } catch (IOException e) { + throw new UncheckedIOException("Error while rewriting query for Lucene pushable filter", e); + } + addToLucenePushableFilters(queryBuilder); + return true; + } + } + return false; + } + + private boolean applyAsLeftRightBinaryComparison( + Expression expr, + List matchFields, + Page inputPage, + ClusterService clusterService, + Warnings warnings + ) { + if (expr instanceof EsqlBinaryComparison binaryComparison + && binaryComparison.left() instanceof Attribute leftAttribute + && binaryComparison.right() instanceof Attribute rightAttribute) { + // the left side comes from the page that was sent to the lookup node + // the right side is the field from the lookup index + // check if the left side is in the matchFields + // if it is its corresponding page is the corresponding number in inputPage + Block block = null; + DataType dataType = null; + for (int i = 0; i < matchFields.size(); i++) { + if (matchFields.get(i).fieldName().equals(leftAttribute.name())) { + block = inputPage.getBlock(i); + dataType = matchFields.get(i).type(); + break; + } + } + MappedFieldType rightFieldType = context.getFieldType(rightAttribute.name()); + if (block != null && rightFieldType != null && dataType != null) { + // special handle Equals operator + // TermQuery is faster than BinaryComparisonQueryList, as it does less work per row + // so here we reuse the existing logic from field based join to build a termQueryList for Equals + if (binaryComparison instanceof Equals) { + QueryList termQueryForEquals = termQueryList(rightFieldType, context, aliasFilter, block, dataType).onlySingleValues( + warnings, + "LOOKUP JOIN encountered multi-value" + ); + queryLists.add(termQueryForEquals); } else { - throw new IllegalStateException( - "Only field from the left dataset are supported on the left of the join on condition but got: " + expr + queryLists.add( + new BinaryComparisonQueryList( + rightFieldType, + context, + block, + binaryComparison, + clusterService, + aliasFilter, + warnings + ) ); } - } else { - // we only support binary comparisons in the join on conditions - throw new IllegalStateException("Only binary comparisons are supported in join ON conditions, but got: " + expr); + return true; } } + return false; } - private void addToPreJoinFilters(QueryBuilder query) { + private void addToLucenePushableFilters(QueryBuilder query) { try { if (query != null) { - preJoinFilters.add(query.toQuery(context)); + lucenePushableFilters.add(query.toQuery(context)); } } catch (IOException e) { - throw new UncheckedIOException("Error while building query for PreJoinFilters filter", e); + throw new UncheckedIOException("Error while building query for Lucene pushable filter", e); } } private void buildPreJoinFilter(PhysicalPlan rightPreJoinPlan, ClusterService clusterService) { if (rightPreJoinPlan instanceof FilterExec filterExec) { List candidateRightHandFilters = Predicates.splitAnd(filterExec.condition()); - LucenePushdownPredicates lucenePushdownPredicates = LucenePushdownPredicates.from( - SearchContextStats.from(List.of(context)), - new EsqlFlags(clusterService.getClusterSettings()) - ); for (Expression filter : candidateRightHandFilters) { if (filter instanceof TranslationAware translationAware) { if (TranslationAware.Translatable.YES.equals(translationAware.translatable(lucenePushdownPredicates))) { - addToPreJoinFilters(translationAware.asQuery(lucenePushdownPredicates, TRANSLATOR_HANDLER).toQueryBuilder()); + addToLucenePushableFilters(translationAware.asQuery(lucenePushdownPredicates, TRANSLATOR_HANDLER).toQueryBuilder()); } } // If the filter is not translatable we will not apply it for now @@ -268,7 +281,7 @@ public Query getQuery(int position) { builder.add(q, BooleanClause.Occur.FILTER); } // also attach the pre-join filter if it exists - for (Query preJoinFilter : preJoinFilters) { + for (Query preJoinFilter : lucenePushableFilters) { builder.add(preJoinFilter, BooleanClause.Occur.FILTER); } return builder.build(); diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/FullTextFunction.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/FullTextFunction.java index 7ab187fe060fa..6e673164bbf0e 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/FullTextFunction.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/FullTextFunction.java @@ -30,6 +30,7 @@ import org.elasticsearch.xpack.esql.core.tree.Source; import org.elasticsearch.xpack.esql.core.type.DataType; import org.elasticsearch.xpack.esql.core.type.MultiTypeEsField; +import org.elasticsearch.xpack.esql.core.util.Holder; import org.elasticsearch.xpack.esql.evaluator.mapper.EvaluatorMapper; import org.elasticsearch.xpack.esql.expression.function.scalar.convert.AbstractConvertFunction; import org.elasticsearch.xpack.esql.expression.predicate.logical.BinaryLogic; @@ -42,6 +43,7 @@ import org.elasticsearch.xpack.esql.plan.logical.Limit; import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan; import org.elasticsearch.xpack.esql.plan.logical.OrderBy; +import org.elasticsearch.xpack.esql.plan.logical.join.LookupJoin; import org.elasticsearch.xpack.esql.planner.EsPhysicalOperationProviders; import org.elasticsearch.xpack.esql.planner.TranslatorHandler; import org.elasticsearch.xpack.esql.querydsl.query.TranslationAwareExpressionQuery; @@ -188,34 +190,11 @@ public BiConsumer postAnalysisPlanVerification() { private static void checkFullTextQueryFunctions(LogicalPlan plan, Failures failures) { if (plan instanceof Filter f) { Expression condition = f.condition(); - - if (condition instanceof Score) { - failures.add(fail(condition, "[SCORE] function can't be used in WHERE")); - } - - List.of(QueryString.class, Kql.class).forEach(functionClass -> { - // Check for limitations of QSTR and KQL function. - checkCommandsBeforeExpression( - plan, - condition, - functionClass, - lp -> (lp instanceof Filter || lp instanceof OrderBy || lp instanceof EsRelation), - fullTextFunction -> "[" + fullTextFunction.functionName() + "] " + fullTextFunction.functionType(), - failures - ); - }); - - checkCommandsBeforeExpression( - plan, - condition, - FullTextFunction.class, - lp -> (lp instanceof Limit == false) && (lp instanceof Aggregate == false), - m -> "[" + m.functionName() + "] " + m.functionType(), - failures - ); - checkFullTextFunctionsParents(condition, failures); + checkFullTextQueryFunctionForCondition(plan, failures, condition, false); } else if (plan instanceof Aggregate agg) { checkFullTextFunctionsInAggs(agg, failures); + } else if (plan instanceof LookupJoin lookupJoin) { + checkFullTextQueryFunctionForCondition(plan, failures, lookupJoin.config().joinOnConditions(), true); } else { List scoredFTFs = new ArrayList<>(); plan.forEachExpression(Score.class, scoreFunction -> { @@ -238,6 +217,43 @@ private static void checkFullTextQueryFunctions(LogicalPlan plan, Failures failu } } + private static void checkFullTextQueryFunctionForCondition( + LogicalPlan plan, + Failures failures, + Expression condition, + boolean isLookupJoinOnCondition + ) { + if (condition == null) { + return; + } + if (condition instanceof Score) { + failures.add(fail(condition, "[SCORE] function can't be used in WHERE")); + } + if (isLookupJoinOnCondition == false) { + List.of(QueryString.class, Kql.class).forEach(functionClass -> { + // Check for limitations of QSTR and KQL function. + checkCommandsBeforeExpression( + plan, + condition, + functionClass, + lp -> (lp instanceof Filter || lp instanceof OrderBy || lp instanceof EsRelation), + fullTextFunction -> "[" + fullTextFunction.functionName() + "] " + fullTextFunction.functionType(), + failures + ); + }); + } + + checkCommandsBeforeExpression( + plan, + condition, + FullTextFunction.class, + lp -> (lp instanceof Limit == false) && (lp instanceof Aggregate == false), + m -> "[" + m.functionName() + "] " + m.functionType(), + failures + ); + checkFullTextFunctionsParents(condition, failures); + } + private static void checkScoreFunction(LogicalPlan plan, Failures failures, Score scoreFunction) { checkCommandsBeforeExpression( plan, @@ -341,6 +357,11 @@ private static FullTextFunction forEachFullTextFunctionParent(Expression conditi } public static void fieldVerifier(LogicalPlan plan, FullTextFunction function, Expression field, Failures failures) { + // Only run the check if the current node contains the full-text function + // This is to avoid running the check multiple times in the same plan + if (isInCurrentNode(plan, function) == false) { + return; + } var fieldAttribute = fieldAsFieldAttribute(field); if (fieldAttribute == null) { plan.forEachExpression(function.getClass(), m -> { @@ -357,6 +378,12 @@ public static void fieldVerifier(LogicalPlan plan, FullTextFunction function, Ex } }); } else { + if (plan instanceof LookupJoin) { + // Full Text Functions are allowed in LOOKUP JOIN ON conditions + // We are only running this code for the node containing the Full Text Function + // So if it is a Lookup Join we know the function is in the join on condition + return; + } // Traverse the plan to find the EsRelation outputting the field plan.forEachDown(p -> { if (p instanceof EsRelation esRelation && esRelation.indexMode() != IndexMode.STANDARD) { @@ -428,4 +455,17 @@ public static FieldAttribute fieldAsFieldAttribute(Expression field) { public void postOptimizationVerification(Failures failures) { resolveTypeQuery(query(), sourceText(), forPostOptimizationValidation(query(), failures)); } + + /** + * Check if the full-text function exists only in the current node (not in child nodes) + */ + private static boolean isInCurrentNode(LogicalPlan plan, FullTextFunction function) { + final Holder found = new Holder<>(false); + plan.forEachExpression(FullTextFunction.class, ftf -> { + if (ftf == function) { + found.set(true); + } + }); + return found.get(); + } } diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/Predicates.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/Predicates.java index b9a58e82a2349..10f4f6a8a8500 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/Predicates.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/Predicates.java @@ -9,6 +9,7 @@ import org.elasticsearch.core.Tuple; import org.elasticsearch.xpack.esql.core.expression.Expression; import org.elasticsearch.xpack.esql.core.expression.Literal; +import org.elasticsearch.xpack.esql.core.tree.Source; import org.elasticsearch.xpack.esql.expression.predicate.logical.And; import org.elasticsearch.xpack.esql.expression.predicate.logical.Or; @@ -49,6 +50,10 @@ public static Expression combineAnd(List exps) { return combine(exps, (l, r) -> new And(l.source(), l, r)); } + public static Expression combineAndWithSource(List exps, Source source) { + return combine(exps, (l, r) -> new And(source, l, r)); + } + /** * Build a binary 'pyramid' from the given list: * 
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/LogicalPlanBuilder.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/LogicalPlanBuilder.java
index dbd18446e748d..ec987ecf46fd9 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/LogicalPlanBuilder.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/LogicalPlanBuilder.java
@@ -46,6 +46,7 @@
 import org.elasticsearch.xpack.esql.expression.UnresolvedNamePattern;
 import org.elasticsearch.xpack.esql.expression.function.UnresolvedFunction;
 import org.elasticsearch.xpack.esql.expression.predicate.Predicates;
+import org.elasticsearch.xpack.esql.expression.predicate.logical.BinaryLogic;
 import org.elasticsearch.xpack.esql.expression.predicate.logical.Not;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.Equals;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.EsqlBinaryComparison;
@@ -699,7 +700,14 @@ public PlanFactory visitJoinCommand(EsqlBaseParser.JoinCommandContext ctx) {
             if (hasRemotes && EsqlCapabilities.Cap.ENABLE_LOOKUP_JOIN_ON_REMOTE.isEnabled() == false) {
                 throw new ParsingException(source, "remote clusters are not supported with LOOKUP JOIN");
             }
-            return new LookupJoin(source, p, right, joinInfo.joinFields(), hasRemotes, Predicates.combineAnd(joinInfo.joinExpressions()));
+            return new LookupJoin(
+                source,
+                p,
+                right,
+                joinInfo.joinFields(),
+                hasRemotes,
+                Predicates.combineAndWithSource(joinInfo.joinExpressions(), source(condition))
+            );
         };
     }
 
@@ -713,7 +721,7 @@ public JoinInfo visitJoinCondition(EsqlBaseParser.JoinConditionContext ctx) {
         }
 
         // inspect the first expression to determine the type of join (field-based or expression-based)
-        boolean isFieldBased = expressions.get(0) instanceof UnresolvedAttribute;
+        boolean isFieldBased = expressions.get(0) instanceof UnresolvedAttribute || expressions.get(0) instanceof Literal;
 
         if (isFieldBased) {
             return processFieldBasedJoin(expressions);
@@ -762,26 +770,53 @@ private JoinInfo processExpressionBasedJoin(List expressions, EsqlBa
         }
         expressions = Predicates.splitAnd(expressions.get(0));
         for (var f : expressions) {
-            addJoinExpression(f, joinFields, joinExpressions);
+            addJoinExpression(f, joinFields, joinExpressions, ctx);
+        }
+        if (joinFields.isEmpty()) {
+            throw new ParsingException(
+                source(ctx),
+                "JOIN ON clause with expressions must contain at least one condition relating the left index and the lookup index"
+            );
         }
         return new JoinInfo(joinFields, joinExpressions);
     }
 
-    private void addJoinExpression(Expression exp, List joinFields, List joinExpressions) {
+    private void addJoinExpression(
+        Expression exp,
+        List joinFields,
+        List joinExpressions,
+        EsqlBaseParser.JoinConditionContext ctx
+    ) {
         exp = handleNegationOfEquals(exp);
+        if (containsBareFieldsInBooleanExpression(exp)) {
+            throw new ParsingException(
+                source(ctx),
+                "JOIN ON clause only supports fields or AND of Binary Expressions at the moment, found [{}]",
+                exp.sourceText()
+            );
+        }
         if (exp instanceof EsqlBinaryComparison comparison
             && comparison.left() instanceof UnresolvedAttribute left
             && comparison.right() instanceof UnresolvedAttribute right) {
             joinFields.add(left);
             joinFields.add(right);
-            joinExpressions.add(exp);
-        } else {
-            throw new ParsingException(
-                exp.source(),
-                "JOIN ON clause only supports fields or AND of Binary Expressions at the moment, found [{}]",
-                exp.sourceText()
-            );
         }
+        joinExpressions.add(exp);
+    }
+
+    private boolean containsBareFieldsInBooleanExpression(Expression expression) {
+        if (expression instanceof UnresolvedAttribute) {
+            return true; // This is a bare field
+        }
+        if (expression instanceof EsqlBinaryComparison) {
+            return false; // This is a binary comparison, not a bare field
+        }
+        if (expression instanceof BinaryLogic binaryLogic) {
+            // Check if either side contains bare fields
+            return containsBareFieldsInBooleanExpression(binaryLogic.left()) || containsBareFieldsInBooleanExpression(binaryLogic.right());
+        }
+        // For other expression types (functions, constants, etc.), they are not bare fields
+        return false;
     }
 
     private void validateJoinFields(List joinFields) {
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/ParsingTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/ParsingTests.java
index 899d3d0647efa..5ad285ff2fdda 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/ParsingTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/ParsingTests.java
@@ -133,22 +133,92 @@ public void testTooBigQuery() {
     public void testJoinOnConstant() {
         assumeTrue(
             "requires LOOKUP JOIN ON boolean expression capability",
-            EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION.isEnabled()
+            EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION_V2.isEnabled()
         );
         assertEquals(
-            "1:55: JOIN ON clause only supports fields or AND of Binary Expressions at the moment, found [123]",
+            "1:55: JOIN ON clause must be a comma separated list of fields or a single expression, found [123]",
             error("row languages = 1, gender = \"f\" | lookup join test on 123")
         );
         assertEquals(
-            "1:55: JOIN ON clause only supports fields or AND of Binary Expressions at the moment, found [\"abc\"]",
+            "1:55: JOIN ON clause must be a comma separated list of fields or a single expression, found [\"abc\"]",
             error("row languages = 1, gender = \"f\" | lookup join test on \"abc\"")
         );
         assertEquals(
-            "1:55: JOIN ON clause only supports fields or AND of Binary Expressions at the moment, found [false]",
+            "1:55: JOIN ON clause must be a comma separated list of fields or a single expression, found [false]",
             error("row languages = 1, gender = \"f\" | lookup join test on false")
         );
     }
 
+    public void testLookupJoinExpressionMixed() {
+        assumeTrue("requires LOOKUP JOIN capability", EsqlCapabilities.Cap.JOIN_LOOKUP_V12.isEnabled());
+        assumeTrue(
+            "requires LOOKUP JOIN ON boolean expression capability",
+            EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION_V2.isEnabled()
+        );
+        String queryString = """
+            from test
+            | rename languages as languages_left
+            | lookup join languages_lookup ON languages_left == language_code or salary > 1000
+            """;
+
+        assertEquals(
+            "3:32: JOIN ON clause with expressions must contain at least one condition relating the left index and the lookup index",
+            error(queryString)
+        );
+    }
+
+    public void testLookupJoinExpressionOnlyRightFilter() {
+        assumeTrue("requires LOOKUP JOIN capability", EsqlCapabilities.Cap.JOIN_LOOKUP_V12.isEnabled());
+        assumeTrue(
+            "requires LOOKUP JOIN ON boolean expression capability",
+            EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION_V2.isEnabled()
+        );
+        String queryString = """
+            from test
+            | rename languages as languages_left
+            | lookup join languages_lookup ON salary > 1000
+            """;
+
+        assertEquals(
+            "3:32: JOIN ON clause with expressions must contain at least one condition relating the left index and the lookup index",
+            error(queryString)
+        );
+    }
+
+    public void testLookupJoinExpressionFieldBasePlusRightFilterAnd() {
+        assumeTrue("requires LOOKUP JOIN capability", EsqlCapabilities.Cap.JOIN_LOOKUP_V12.isEnabled());
+        assumeTrue(
+            "requires LOOKUP JOIN ON boolean expression capability",
+            EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION_V2.isEnabled()
+        );
+        String queryString = """
+            from test
+            | lookup join languages_lookup ON languages and salary > 1000
+            """;
+
+        assertEquals(
+            "2:32: JOIN ON clause only supports fields or AND of Binary Expressions at the moment, found [languages]",
+            error(queryString)
+        );
+    }
+
+    public void testLookupJoinExpressionFieldBasePlusRightFilterComma() {
+        assumeTrue("requires LOOKUP JOIN capability", EsqlCapabilities.Cap.JOIN_LOOKUP_V12.isEnabled());
+        assumeTrue(
+            "requires LOOKUP JOIN ON boolean expression capability",
+            EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION_V2.isEnabled()
+        );
+        String queryString = """
+            from test
+            | lookup join languages_lookup ON languages, salary > 1000
+            """;
+
+        assertEquals(
+            "2:46: JOIN ON clause must be a comma separated list of fields or a single expression, found [salary > 1000]",
+            error(queryString)
+        );
+    }
+
     public void testJoinTwiceOnTheSameField() {
         assertEquals(
             "1:66: JOIN ON clause does not support multiple fields with the same name, found multiple instances of [languages]",
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/VerifierTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/VerifierTests.java
index 2a139d820ed3a..6965b69388aa9 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/VerifierTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/VerifierTests.java
@@ -2253,6 +2253,21 @@ public void testLookupJoinExpressionAmbiguousRight() {
         );
     }
 
+    public void testLookupJoinExpressionRightNotPushable() {
+        assumeTrue("requires LOOKUP JOIN capability", EsqlCapabilities.Cap.JOIN_LOOKUP_V12.isEnabled());
+        assumeTrue(
+            "requires LOOKUP JOIN ON boolean expression capability",
+            EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION_V2.isEnabled()
+        );
+        String queryString = """
+            from test
+            | rename languages as languages_left
+            | lookup join languages_lookup ON languages_left == language_code and abs(salary) > 1000
+            """;
+
+        assertEquals("3:71: Unsupported join filter expression:abs(salary) > 1000", error(queryString));
+    }
+
     public void testLookupJoinExpressionAmbiguousLeft() {
         assumeTrue(
             "requires LOOKUP JOIN ON boolean expression capability",
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexOperatorTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexOperatorTests.java
index c342377e7894f..e02ef6110e5c4 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexOperatorTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexOperatorTests.java
@@ -108,6 +108,7 @@ public class LookupFromIndexOperatorTests extends AsyncOperatorTestCase {
     private final ThreadPool threadPool = threadPool();
     private final Directory lookupIndexDirectory = newDirectory();
     private final List releasables = new ArrayList<>();
+    private final boolean applyRightFilterAsJoinOnFilter;
     private int numberOfJoinColumns; // we only allow 1 or 2 columns due to simpleInput() implementation
     private EsqlBinaryComparison.BinaryComparisonOperation operation;
 
@@ -129,6 +130,7 @@ public static Iterable parametersFactory() {
     public LookupFromIndexOperatorTests(EsqlBinaryComparison.BinaryComparisonOperation operation) {
         super();
         this.operation = operation;
+        this.applyRightFilterAsJoinOnFilter = randomBoolean();
     }
 
     @Before
@@ -248,6 +250,7 @@ protected Operator.OperatorFactory simple(SimpleOptions options) {
             matchFields.add(new MatchConfig(matchField, i, inputDataType));
         }
         Expression joinOnExpression = null;
+        FragmentExec rightPlanWithOptionalPreJoinFilter = buildLessThanFilter(LESS_THAN_VALUE);
         if (operation != null) {
             List conditions = new ArrayList<>();
             for (int i = 0; i < numberOfJoinColumns; i++) {
@@ -265,6 +268,13 @@ protected Operator.OperatorFactory simple(SimpleOptions options) {
                 );
                 conditions.add(operation.buildNewInstance(Source.EMPTY, left, right));
             }
+            if (applyRightFilterAsJoinOnFilter) {
+                if (rightPlanWithOptionalPreJoinFilter instanceof FragmentExec fragmentExec
+                    && fragmentExec.fragment() instanceof Filter filterPlan) {
+                    conditions.add(filterPlan.condition());
+                    rightPlanWithOptionalPreJoinFilter = null;
+                }
+            }
             joinOnExpression = Predicates.combineAnd(conditions);
         }
 
@@ -278,7 +288,7 @@ protected Operator.OperatorFactory simple(SimpleOptions options) {
             lookupIndex,
             loadFields,
             Source.EMPTY,
-            buildLessThanFilter(LESS_THAN_VALUE),
+            rightPlanWithOptionalPreJoinFilter,
             joinOnExpression
         );
     }
@@ -321,25 +331,41 @@ protected Matcher expectedToStringOfSimple() {
             // match_field=match_left (index first, then suffix)
             sb.append("input_type=LONG match_field=match").append(i).append(suffix).append(" inputChannel=").append(i).append(" ");
         }
-        // Accept either the legacy physical plan rendering (FilterExec/EsQueryExec) or the new FragmentExec rendering
-        sb.append("right_pre_join_plan=(?:");
-        // Legacy pattern
-        sb.append("FilterExec\\[lint\\{f}#\\d+ < ")
-            .append(LESS_THAN_VALUE)
-            .append(
-                "\\[INTEGER]]\\n\\\\_EsQueryExec\\[test], indexMode\\[lookup],\\s*(?:query\\[\\]|\\[\\])?,?\\s*"
-                    + "limit\\[\\],?\\s*sort\\[(?:\\[\\])?\\]\\s*estimatedRowSize\\[null\\]\\s*queryBuilderAndTags \\[(?:\\[\\]\\])\\]"
-            );
-        sb.append("|");
-        // New FragmentExec pattern - match the actual output format
-        sb.append("FragmentExec\\[filter=null, estimatedRowSize=\\d+, reducer=\\[\\], fragment=\\[<>\\n")
-            .append("Filter\\[lint\\{f}#\\d+ < ")
-            .append(LESS_THAN_VALUE)
-            .append("\\[INTEGER]]\\n")
-            .append("\\\\_EsRelation\\[test]\\[LOOKUP]\\[\\]<>\\]\\]");
-        sb.append(")");
+
+        if (applyRightFilterAsJoinOnFilter && operation != null) {
+            // When applyRightFilterAsJoinOnFilter is true, right_pre_join_plan should be null
+            sb.append("right_pre_join_plan=null");
+        } else {
+            // Accept either the legacy physical plan rendering (FilterExec/EsQueryExec) or the new FragmentExec rendering
+            sb.append("right_pre_join_plan=(?:");
+            // Legacy pattern
+            sb.append("FilterExec\\[lint\\{f}#\\d+ < ")
+                .append(LESS_THAN_VALUE)
+                .append(
+                    "\\[INTEGER]]\\n\\\\_EsQueryExec\\[test], indexMode\\[lookup],\\s*(?:query\\[\\]|\\[\\])?,?\\s*"
+                        + "limit\\[\\],?\\s*sort\\[(?:\\[\\])?\\]\\s*estimatedRowSize\\[null\\]\\s*queryBuilderAndTags \\[(?:\\[\\]\\])\\]"
+                );
+            sb.append("|");
+            // New FragmentExec pattern - match the actual output format
+            sb.append("FragmentExec\\[filter=null, estimatedRowSize=\\d+, reducer=\\[\\], fragment=\\[<>\\n")
+                .append("Filter\\[lint\\{f}#\\d+ < ")
+                .append(LESS_THAN_VALUE)
+                .append("\\[INTEGER]]\\n")
+                .append("\\\\_EsRelation\\[test]\\[LOOKUP]\\[\\]<>\\]\\]");
+            sb.append(")");
+        }
+
         // Accept join_on_expression=null or a valid join predicate
-        sb.append(" join_on_expression=(null|match\\d+left [=!<>]+ match\\d+right( AND match\\d+left [=!<>]+ match\\d+right)*|)\\]");
+        if (applyRightFilterAsJoinOnFilter && operation != null) {
+            // When applyRightFilterAsJoinOnFilter is true and operation is not null, the join expression includes the filter condition
+            sb.append(
+                " join_on_expression=(match\\d+left [=!<>]+ match\\d+right( "
+                    + "AND match\\d+left [=!<>]+ match\\d+right)* AND lint\\{f}#\\d+ < "
+            ).append(LESS_THAN_VALUE).append("\\[INTEGER]|)\\]");
+        } else {
+            // Standard pattern for other cases
+            sb.append(" join_on_expression=(null|match\\d+left [=!<>]+ match\\d+right( AND match\\d+left [=!<>]+ match\\d+right)*|)\\]");
+        }
         return matchesPattern(sb.toString());
     }
 
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java
index ab21c700b41be..6100e4145ba21 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java
@@ -5367,6 +5367,49 @@ public void testPlanSanityCheckWithBinaryPlans() {
         assertThat(e.getMessage(), containsString(" optimized incorrectly due to missing references from right hand side [language_code"));
     }
 
+    /**
+     * Expected
+     * 
{@code
+     * Limit[1000[INTEGER],true]
+     * \_Join[LEFT,[languages{f}#8],[language_code{f}#16],languages{f}#8 == language_code{f}#16 AND language_name{f}#17 == English
+     * [KEYWORD]]
+     *   |_Limit[1000[INTEGER],false]
+     *   | \_EsRelation[test][_meta_field{f}#11, emp_no{f}#5, first_name{f}#6, ge..]
+     *   \_EsRelation[languages_lookup][LOOKUP][language_code{f}#16, language_name{f}#17]
+     * }

+     */
+    public void testLookupJoinRightFilter() {
+        assumeTrue("Requires LOOKUP JOIN", EsqlCapabilities.Cap.JOIN_LOOKUP_V12.isEnabled());
+
+        var plan = optimizedPlan("""
+              FROM test
+            | LOOKUP JOIN languages_lookup ON languages == language_code and language_name == "English"
+            """);
+
+        var upperLimit = asLimit(plan, 1000, true);
+        var join = as(upperLimit.child(), Join.class);
+        assertEquals("ON languages == language_code and language_name == \"English\"", join.config().joinOnConditions().toString());
+        var limitPastJoin = asLimit(join.left(), 1000, false);
+        as(limitPastJoin.child(), EsRelation.class);
+        as(join.right(), EsRelation.class);
+    }
+
+    public void testLookupJoinRightFilterMatch() {
+        assumeTrue("Requires LOOKUP JOIN", EsqlCapabilities.Cap.JOIN_LOOKUP_V12.isEnabled());
+
+        var plan = optimizedPlan("""
+              FROM test
+            | LOOKUP JOIN languages_lookup ON languages == language_code and MATCH(language_name,"English")
+            """);
+
+        var upperLimit = asLimit(plan, 1000, true);
+        var join = as(upperLimit.child(), Join.class);
+        assertEquals("ON languages == language_code and MATCH(language_name,\"English\")", join.config().joinOnConditions().toString());
+        var limitPastJoin = asLimit(join.left(), 1000, false);
+        as(limitPastJoin.child(), EsRelation.class);
+        as(join.right(), EsRelation.class);
+    }
+
     // https://github.com/elastic/elasticsearch/issues/104995
     public void testNoWrongIsNotNullPruning() {
         var plan = optimizedPlan("""
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/StatementParserTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/StatementParserTests.java
index d92d1b21a98e9..df0e7dd81e8a0 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/StatementParserTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/StatementParserTests.java
@@ -3508,6 +3508,28 @@ public void testInvalidLookupJoinOnClause() {
             "JOIN ON clause only supports fields or AND of Binary Expressions at the moment, found"
         );
 
+        expectError(
+            "FROM test  | LOOKUP JOIN test2 ON "
+                + singleExpressionJoinClause()
+                + " AND ("
+                + randomIdentifier()
+                + " OR "
+                + singleExpressionJoinClause()
+                + ")",
+            "JOIN ON clause only supports fields or AND of Binary Expressions at the moment, found"
+        );
+
+        expectError(
+            "FROM test  | LOOKUP JOIN test2 ON "
+                + singleExpressionJoinClause()
+                + " AND ("
+                + randomIdentifier()
+                + "OR"
+                + randomIdentifier()
+                + ")",
+            "JOIN ON clause only supports fields or AND of Binary Expressions at the moment, found"
+        );
+
         expectError(
             "FROM test  | LOOKUP JOIN test2 ON " + randomIdentifier() + " AND " + randomIdentifier(),
             "JOIN ON clause only supports fields or AND of Binary Expressions at the moment, found"
@@ -5047,7 +5069,7 @@ public void testMixedSingleDoubleParams() {
                     expectError(
                         LoggerMessageFormat.format(null, "from test | " + command, param1, param2, param3),
                         List.of(paramAsConstant("f1", "f1"), paramAsConstant("f2", "f2"), paramAsConstant("f3", "f3")),
-                        "JOIN ON clause only supports fields or AND of Binary Expressions at the moment"
+                        "line 1:33: JOIN ON clause must be a comma separated list of fields or a single expression, found [?f1]"
                     );
 
                 }