ESQL: Add support for Full Text Functions and Lucene Pushable Predicates for Lookup Join (elastic#136104)

• Added Full Text Functions Support in Lookup Join Expressions: Extended lookup join functionality to support full-text functions like MATCH() as part of join conditions, enabling text search capabilities within lookup operations.
• Enhanced Lucene Pushable Filters Support. Implemented support for any Lucene pushable filters that contain only fields from the right side of the lookup join, allowing for more complex filtering conditions. At least one condition relating the left and right side of the Lookup Join is still required.

This commit may include content that was generated or assisted by GenAI tools Cursor, Gemini CLI and/or Github Copilot

Author: julian-elastic

docs/changelog/136104.yaml

@@ -0,0 +1,5 @@
+pr: 136104
+summary: Add support for Full Text Functions and Lucene pushable conditions on fields from the Lookup Index for Lookup Join
+area: ES|QL
+type: enhancement
+issues: [ ]

server/src/main/resources/transport/definitions/referable/esql_lookup_join_full_text_function.csv

@@ -0,0 +1 @@
+9201000

server/src/main/resources/transport/upper_bounds/9.3.csv

@@ -1 +1 @@
-inference_cached_tokens,9200000
+esql_lookup_join_full_text_function,9201000

test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java

@@ -861,11 +861,22 @@ private Map<String, Object> 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));

x-pack/plugin/esql/qa/security/src/javaRestTest/java/org/elasticsearch/xpack/esql/EsqlSecurityIT.java

@@ -828,7 +828,7 @@ private void testLookupJoinFieldLevelSecurityHelper(boolean useExpressionJoin) t
         ResponseException error = expectThrows(ResponseException.class, () -> runESQLCommand("fls_user4_1", query));
         assertThat(error.getResponse().getStatusLine().getStatusCode(), equalTo(HttpStatus.SC_BAD_REQUEST));
         if (useExpressionJoin) {
-            assertThat(error.getMessage(), containsString("Unsupported join filter expression:value_left == value"));
+            assertThat(error.getMessage(), containsString("Unknown column [value], did you mean [value_left]?"));
         } else {
             assertThat(error.getMessage(), containsString("Unknown column [value] in right side of join"));
         }
@@ -902,7 +902,7 @@ private void testLookupJoinFieldLevelSecurityOnAliasHelper(boolean useExpression
         ResponseException error = expectThrows(ResponseException.class, () -> runESQLCommand("fls_user4_1_alias", query));
         assertThat(error.getResponse().getStatusLine().getStatusCode(), equalTo(HttpStatus.SC_BAD_REQUEST));
         if (useExpressionJoin) {
-            assertThat(error.getMessage(), containsString("Unsupported join filter expression:value_left == value"));
+            assertThat(error.getMessage(), containsString("Unknown column [value], did you mean [value_left]?"));
         } else {
             assertThat(error.getMessage(), containsString("Unknown column [value] in right side of join"));
         }

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/CsvTestsDataLoader.java

@@ -149,7 +149,9 @@ public class CsvTestsDataLoader {
     private static final TestDataset DATE_NANOS_UNION_TYPES = new TestDataset("date_nanos_union_types");
     private static final TestDataset COUNTRIES_BBOX = new TestDataset("countries_bbox");
     private static final TestDataset COUNTRIES_BBOX_WEB = new TestDataset("countries_bbox_web");
-    private static final TestDataset AIRPORT_CITY_BOUNDARIES = new TestDataset("airport_city_boundaries");
+    private static final TestDataset AIRPORT_CITY_BOUNDARIES = new TestDataset("airport_city_boundaries").withSetting(
+        "lookup-settings.json"
+    );
     private static final TestDataset CARTESIAN_MULTIPOLYGONS = new TestDataset("cartesian_multipolygons");
     private static final TestDataset CARTESIAN_MULTIPOLYGONS_NO_DOC_VALUES = new TestDataset("cartesian_multipolygons_no_doc_values")
         .withData("cartesian_multipolygons.csv");

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java

@@ -62,9 +62,9 @@
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xcontent.json.JsonXContent;
 import org.elasticsearch.xpack.esql.action.EsqlQueryResponse;
-import org.elasticsearch.xpack.esql.analysis.AnalyzerContext;
 import org.elasticsearch.xpack.esql.analysis.AnalyzerSettings;
 import org.elasticsearch.xpack.esql.analysis.EnrichResolution;
+import org.elasticsearch.xpack.esql.analysis.MutableAnalyzerContext;
 import org.elasticsearch.xpack.esql.analysis.Verifier;
 import org.elasticsearch.xpack.esql.core.expression.Alias;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
@@ -449,7 +449,7 @@ public static TransportVersion randomMinimumVersion() {
     }
 
     // TODO: make this even simpler, remove the enrichResolution for tests that do not require it (most tests)
-    public static AnalyzerContext testAnalyzerContext(
+    public static MutableAnalyzerContext testAnalyzerContext(
         Configuration configuration,
         EsqlFunctionRegistry functionRegistry,
         Map<IndexPattern, IndexResolution> indexResolutions,
@@ -462,15 +462,15 @@ public static AnalyzerContext testAnalyzerContext(
     /**
      * Analyzer context for a random (but compatible) minimum transport version.
      */
-    public static AnalyzerContext testAnalyzerContext(
+    public static MutableAnalyzerContext testAnalyzerContext(
         Configuration configuration,
         EsqlFunctionRegistry functionRegistry,
         Map<IndexPattern, IndexResolution> indexResolutions,
         Map<String, IndexResolution> lookupResolution,
         EnrichResolution enrichResolution,
         InferenceResolution inferenceResolution
     ) {
-        return new AnalyzerContext(
+        return new MutableAnalyzerContext(
             configuration,
             functionRegistry,
             indexResolutions,

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/analysis/MutableAnalyzerContext.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.analysis;
+
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.TransportVersionUtils;
+import org.elasticsearch.xpack.esql.expression.function.EsqlFunctionRegistry;
+import org.elasticsearch.xpack.esql.index.IndexResolution;
+import org.elasticsearch.xpack.esql.inference.InferenceResolution;
+import org.elasticsearch.xpack.esql.plan.IndexPattern;
+import org.elasticsearch.xpack.esql.session.Configuration;
+
+import java.util.Map;
+
+/**
+ * A mutable version of AnalyzerContext that allows temporarily changing the transport version.
+ * This is useful for testing scenarios where different transport versions need to be tested.
+ */
+public class MutableAnalyzerContext extends AnalyzerContext {
+    private TransportVersion currentVersion;
+
+    public MutableAnalyzerContext(
+        Configuration configuration,
+        EsqlFunctionRegistry functionRegistry,
+        Map<IndexPattern, IndexResolution> indexResolution,
+        Map<String, IndexResolution> lookupResolution,
+        EnrichResolution enrichResolution,
+        InferenceResolution inferenceResolution,
+        TransportVersion minimumVersion
+    ) {
+        super(configuration, functionRegistry, indexResolution, lookupResolution, enrichResolution, inferenceResolution, minimumVersion);
+        this.currentVersion = minimumVersion;
+    }
+
+    @Override
+    public TransportVersion minimumVersion() {
+        return currentVersion;
+    }
+
+    /**
+     * Temporarily set the transport version to a random version between the passed-in version and the latest,
+     * and return an AutoCloseable to restore it.
+     * Usage:
+     * try (var restore = context.setTemporaryTransportVersionOnOrAfter(minVersion)) {...}
+     */
+    public RestoreTransportVersion setTemporaryTransportVersionOnOrAfter(TransportVersion minVersion) {
+        TransportVersion oldVersion = this.currentVersion;
+        // Set to a random version between minVersion and current
+        this.currentVersion = TransportVersionUtils.randomVersionBetween(ESTestCase.random(), minVersion, TransportVersion.current());
+        return new RestoreTransportVersion(oldVersion);
+    }
+
+    /**
+     * AutoCloseable that restores the original transport version when closed.
+     */
+    public class RestoreTransportVersion implements AutoCloseable {
+        private final TransportVersion originalVersion;
+
+        private RestoreTransportVersion(TransportVersion originalVersion) {
+            this.originalVersion = originalVersion;
+        }
+
+        @Override
+        public void close() {
+            MutableAnalyzerContext.this.currentVersion = originalVersion;
+        }
+    }
+}