ESQL: Add support for Full Text Functions and Lucene Pushable Predicates for Lookup Join

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;
+        }
+    }
+}