Initial support for TEXT fields in LOOKUP JOIN condition

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/QueryList.java

@@ -70,7 +70,11 @@ public static QueryList rawTermQueryList(MappedFieldType field, SearchExecutionC
             }
             case BYTES_REF -> offset -> {
                 BytesRefBlock bytesRefBlock = (BytesRefBlock) block;
-                return bytesRefBlock.getBytesRef(offset, new BytesRef());
+                BytesRef value = bytesRefBlock.getBytesRef(offset, new BytesRef());
+                if (field.typeName().equals("text")) {
+                    throw new IllegalArgumentException("Cannot perform LOOKUP JOIN on TEXT fields");
+                }
+                return value;
             };
             case DOUBLE -> {
                 DoubleBlock doubleBlock = ((DoubleBlock) block);

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

@@ -462,7 +462,7 @@ language_code:integer | language_name:keyword       | country:keyword
 ;
 
 ###########################################################################
-# nested filed join behavior with languages_nested_fields index
+# nested field join behavior with languages_nested_fields index
 ###########################################################################
 
 joinOnNestedField
@@ -510,6 +510,34 @@ language.id:integer | language.name:text | language.name.keyword:keyword
 1                   | English            | English
 ;
 
+joinOnNestedNestedFieldRowImplicitKeyword
+required_capability: join_lookup_v10
+required_capability: lookup_join_text
+
+ROW language.name = "English"
+| LOOKUP JOIN languages_nested_fields ON language.name
+| KEEP language.id, language.name, language.name.keyword
+;
+
+language.id:integer | language.name:keyword | language.name.keyword:keyword
+1                   | English               | English
+;
+
+joinOnNestedNestedFieldRowImplicitKeywords
+required_capability: join_lookup_v10
+required_capability: lookup_join_text
+
+ROW language.name = ["English", "French"]
+| MV_EXPAND language.name
+| LOOKUP JOIN languages_nested_fields ON language.name
+| KEEP language.id, language.name, language.name.keyword, language.code
+;
+
+language.id:integer | language.name:keyword | language.name.keyword:keyword | language.code:keyword
+1                   | English               | English                       | EN
+2                   | French                | French                        | FR
+;
+
 ###############################################
 # Tests with clientips_lookup index
 ###############################################

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

@@ -562,6 +562,11 @@ public enum Cap {
          */
         JOIN_LOOKUP_V10(Build.current().isSnapshot()),
 
+        /**
+         * LOOKUP JOIN with TEXT fields on the right (right side of the join)
+         */
+        LOOKUP_JOIN_TEXT(Build.current().isSnapshot()),
+
         /**
          * Fix for https://github.com/elastic/elasticsearch/issues/117054
          */

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Verifier.java

@@ -858,7 +858,7 @@ private static void checkJoin(LogicalPlan plan, Set<Failure> failures) {
             for (int i = 0; i < config.leftFields().size(); i++) {
                 Attribute leftField = config.leftFields().get(i);
                 Attribute rightField = config.rightFields().get(i);
-                if (leftField.dataType() != rightField.dataType()) {
+                if (leftField.dataType().noText() != rightField.dataType().noText()) {
                     failures.add(
                         fail(
                             leftField,

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexService.java

@@ -80,7 +80,8 @@ protected String getRequiredPrivilege() {
 
     private static void validateTypes(DataType inputDataType, MappedFieldType fieldType) {
         // TODO: consider supporting implicit type conversion as done in ENRICH for some types
-        if (fieldType.typeName().equals(inputDataType.typeName()) == false) {
+        String typeName = fieldType.typeName().equals("text") ? DataType.KEYWORD.typeName() : fieldType.typeName();
+        if (typeName.equals(inputDataType.noText().typeName()) == false) {
             throw new EsqlIllegalArgumentException(
                 "LOOKUP JOIN match and input types are incompatible: match[" + fieldType.typeName() + "], input[" + inputDataType + "]"
             );

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java

@@ -55,9 +55,11 @@
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.Expressions;
+import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
 import org.elasticsearch.xpack.esql.core.expression.NameId;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
+import org.elasticsearch.xpack.esql.core.expression.TypedAttribute;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.util.Holder;
@@ -570,35 +572,49 @@ private PhysicalOperation planLookupJoin(LookupJoinExec join, LocalExecutionPlan
             throw new IllegalArgumentException("can't plan [" + join + "], found index with mode [" + entry.getValue() + "]");
         }
         String indexName = entry.getKey();
-        List<Layout.ChannelAndType> matchFields = new ArrayList<>(join.leftFields().size());
-        for (Attribute m : join.leftFields()) {
-            Layout.ChannelAndType t = source.layout.get(m.id());
-            if (t == null) {
-                throw new IllegalArgumentException("can't plan [" + join + "][" + m + "]");
+        if (join.leftFields().size() != join.rightFields().size()) {
+            throw new IllegalArgumentException("can't plan [" + join + "]: mismatching left and right field count");
+        }
+        List<MatchConfig> matchFields = new ArrayList<>(join.leftFields().size());
+        for (int i = 0; i < join.leftFields().size(); i++) {
+            TypedAttribute left = (TypedAttribute) join.leftFields().get(i);
+            FieldAttribute right = (FieldAttribute) join.rightFields().get(i);
+            Layout.ChannelAndType input = source.layout.get(left.id());
+            if (input == null) {
+                throw new IllegalArgumentException("can't plan [" + join + "][" + left + "]");
             }
-            matchFields.add(t);
+            matchFields.add(new MatchConfig(right, input));
         }
         if (matchFields.size() != 1) {
-            throw new IllegalArgumentException("can't plan [" + join + "]");
+            throw new IllegalArgumentException("can't plan [" + join + "]: multiple join predicates are not supported");
         }
+        // TODO support multiple match fields, and support more than equality predicates
+        MatchConfig matchConfig = matchFields.getFirst();
 
         return source.with(
             new LookupFromIndexOperator.Factory(
                 sessionId,
                 parentTask,
                 context.queryPragmas().enrichMaxWorkers(),
-                matchFields.getFirst().channel(),
+                matchConfig.channel(),
                 lookupFromIndexService,
-                matchFields.getFirst().type(),
+                matchConfig.type(),
                 indexName,
-                join.leftFields().getFirst().name(),
+                matchConfig.fieldName(),
                 join.addedFields().stream().map(f -> (NamedExpression) f).toList(),
                 join.source()
             ),
             layout
         );
     }
 
+    private record MatchConfig(String fieldName, int channel, DataType type) {
+        private MatchConfig(FieldAttribute match, Layout.ChannelAndType input) {
+            // Note, this handles TEXT fields with KEYWORD subfields, and we assume tha has been validated earlier during planning
+            this(match.exactAttribute().name(), input.channel(), input.type());
+        }
+    }
+
     private ExpressionEvaluator.Factory toEvaluator(Expression exp, Layout layout) {
         return EvalMapper.toEvaluator(exp, layout);
     }