ESQL: Fix some more usages of field attribute names in LOOKUP JOIN

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/FieldAttribute.java

@@ -34,8 +34,8 @@
  *
  * To adequately represent e.g. union types, the name of the attribute can be altered because we may have multiple synthetic field
  * attributes that really belong to the same underlying field. For instance, if a multi-typed field is used both as {@code field::string}
- * and {@code field::ip}, we'll generate 2 field attributes called {@code $$field$converted_to$string} and {@code $$field$converted_to$ip}
- * but still referring to the same underlying field.
+ * and {@code field::ip}, we'll generate 2 field attributes called {@code $$field$converted_to$keyword} and {@code $$field$converted_to$ip}
+ * which still refer to the same underlying index field.
  */
 public class FieldAttribute extends TypedAttribute {
 
@@ -162,6 +162,15 @@ public FieldName fieldName() {
         return lazyFieldName;
     }
 
+    /**
+     * The name of the attribute. Can deviate from the field name e.g. in case of union types. For the physical field name, use
+     * {@link FieldAttribute#fieldName()}.
+     */
+    @Override
+    public String name() {
+        return super.name();
+    }
+
     public EsField.Exact getExactInfo() {
         return field.getExactInfo();
     }
@@ -175,7 +184,7 @@ public FieldAttribute exactAttribute() {
     }
 
     private FieldAttribute innerField(EsField type) {
-        return new FieldAttribute(source(), name(), name() + "." + type.getName(), type, nullable(), id(), synthetic());
+        return new FieldAttribute(source(), fieldName().string, name() + "." + type.getName(), type, nullable(), id(), synthetic());
     }
 
     @Override

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

@@ -732,6 +732,28 @@ language.id:integer | language.name:text | language.name.keyword:keyword | langu
 2                   | French             | French                        | FR
 ;
 
+###############################################
+# union type behavior
+###############################################
+
+joinOnMultiTypedMatchFieldCastToInteger
+required_capability: join_lookup_v12
+
+FROM apps, apps_short METADATA _index
+| EVAL language_code = id::integer
+| KEEP _index, language_code
+| WHERE language_code < 3
+| LOOKUP JOIN languages_lookup ON language_code
+| SORT _index ASC, language_code ASC
+;
+
+_index:keyword | language_code:integer | language_name:keyword
+apps           | 1                     | English
+apps           | 2                     | French
+apps_short     | 1                     | English
+apps_short     | 2                     | French
+;
+
 ###############################################
 # Tests with clientips_lookup index
 ###############################################

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

@@ -71,6 +71,7 @@
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
 import org.elasticsearch.xpack.esql.core.expression.Alias;
+import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
@@ -415,8 +416,13 @@ private static Operator extractFieldsOperator(
     ) {
         List<ValuesSourceReaderOperator.FieldInfo> fields = new ArrayList<>(extractFields.size());
         for (NamedExpression extractField : extractFields) {
+            String fieldName = extractField instanceof FieldAttribute fa ? fa.fieldName().string()
+                // Cases for Alias and ReferenceAttribute: only required for ENRICH (Alias in case of ENRICH ... WITH x = field)
+                // (LOOKUP JOIN uses FieldAttributes)
+                : extractField instanceof Alias a ? ((NamedExpression) a.child()).name()
+                : extractField.name();
             BlockLoader loader = shardContext.blockLoader(
-                extractField instanceof Alias a ? ((NamedExpression) a.child()).name() : extractField.name(),
+                fieldName,
                 extractField.dataType() == DataType.UNSUPPORTED,
                 MappedFieldType.FieldExtractPreference.NONE
             );

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

@@ -23,6 +23,7 @@
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
@@ -45,7 +46,7 @@ public record Factory(
         DataType inputDataType,
         String lookupIndexPattern,
         String lookupIndex,
-        String matchField,
+        FieldAttribute.FieldName matchField,
         List<NamedExpression> loadFields,
         Source source
     ) implements OperatorFactory {
@@ -56,7 +57,7 @@ public String describe() {
                 + " input_type="
                 + inputDataType
                 + " match_field="
-                + matchField
+                + matchField.string()
                 + " load_fields="
                 + loadFields
                 + " inputChannel="
@@ -76,7 +77,7 @@ public Operator get(DriverContext driverContext) {
                 inputDataType,
                 lookupIndexPattern,
                 lookupIndex,
-                matchField,
+                matchField.string(),
                 loadFields,
                 source
             );

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/FullTextFunction.java

@@ -396,6 +396,8 @@ protected Map<String, Object> resolvedOptions() throws InvalidArgumentException
         return Map.of();
     }
 
+    // TODO: this should likely be replaced by calls to FieldAttribute#fieldName; the MultiTypeEsField case looks
+    // wrong if `fieldAttribute` is a subfield, e.g. `parent.child` - multiTypeEsField#getName will just return `child`.
     public static String getNameFromFieldAttribute(FieldAttribute fieldAttribute) {
         String fieldName = fieldAttribute.name();
         if (fieldAttribute.field() instanceof MultiTypeEsField multiTypeEsField) {

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

@@ -158,6 +158,9 @@ public AttributeSet rightReferences() {
         return Expressions.references(config().rightFields());
     }
 
+    /**
+     * The output fields obtained from the right child.
+     */
     public List<Attribute> rightOutputFields() {
         AttributeSet leftInputs = left().outputSet();
 

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

@@ -766,10 +766,10 @@ private PhysicalOperation planLookupJoin(LookupJoinExec join, LocalExecutionPlan
         );
     }
 
-    private record MatchConfig(String fieldName, int channel, DataType type) {
+    private record MatchConfig(FieldAttribute.FieldName fieldName, int channel, DataType type) {
         private MatchConfig(FieldAttribute match, Layout.ChannelAndType input) {
             // Note, this handles TEXT fields with KEYWORD subfields
-            this(match.exactAttribute().name(), input.channel(), input.type());
+            this(match.exactAttribute().fieldName(), input.channel(), input.type());
         }
     }
 

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexOperatorTests.java

@@ -56,6 +56,7 @@
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.core.expression.ReferenceAttribute;
 import org.elasticsearch.xpack.esql.core.tree.Source;
@@ -134,7 +135,7 @@ protected Operator.OperatorFactory simple(SimpleOptions options) {
         int inputChannel = 0;
         DataType inputDataType = DataType.LONG;
         String lookupIndex = "idx";
-        String matchField = "match";
+        FieldAttribute.FieldName matchField = new FieldAttribute.FieldName("match");
         List<NamedExpression> loadFields = List.of(
             new ReferenceAttribute(Source.EMPTY, "lkwd", DataType.KEYWORD),
             new ReferenceAttribute(Source.EMPTY, "lint", DataType.INTEGER)