Fix union types in CCS (#128111)

docs/changelog/128111.yaml

@@ -0,0 +1,5 @@
+pr: 128111
+summary: Fix union types in CCS
+area: ES|QL
+type: bug
+issues: []

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/Clusters.java

@@ -56,6 +56,12 @@ public static org.elasticsearch.Version remoteClusterVersion() {
         return prop != null ? org.elasticsearch.Version.fromString(prop) : org.elasticsearch.Version.CURRENT;
     }
 
+    public static org.elasticsearch.Version bwcVersion() {
+        org.elasticsearch.Version local = localClusterVersion();
+        org.elasticsearch.Version remote = remoteClusterVersion();
+        return local.before(remote) ? local : remote;
+    }
+
     private static Version distributionVersion(String key) {
         final String val = System.getProperty(key);
         return val != null ? Version.fromString(val) : Version.CURRENT;

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClustersQueryIT.java

@@ -1058,4 +1058,29 @@ private void clearSkipUnavailable() {
             .setPersistentSettings(settingsBuilder.build())
             .get();
     }
+
+    public void testMultiTypes() throws Exception {
+        Client remoteClient = client(REMOTE_CLUSTER_1);
+        int totalDocs = 0;
+        for (String type : List.of("integer", "long")) {
+            String index = "conflict-index-" + type;
+            assertAcked(remoteClient.admin().indices().prepareCreate(index).setMapping("port", "type=" + type));
+            int numDocs = between(1, 10);
+            for (int i = 0; i < numDocs; i++) {
+                remoteClient.prepareIndex(index).setId(Integer.toString(i)).setSource("port", i).get();
+            }
+            remoteClient.admin().indices().prepareRefresh(index).get();
+            totalDocs += numDocs;
+        }
+        for (String castFunction : List.of("TO_LONG", "TO_INT")) {
+            EsqlQueryRequest request = new EsqlQueryRequest();
+            request.query("FROM *:conflict-index-* | EVAL port=" + castFunction + "(port) | WHERE port is NOT NULL | STATS COUNT(port)");
+            try (EsqlQueryResponse resp = runQuery(request)) {
+                List<List<Object>> values = getValuesList(resp);
+                assertThat(values, hasSize(1));
+                assertThat(values.get(0), hasSize(1));
+                assertThat(values.get(0).get(0), equalTo((long) totalDocs));
+            }
+        }
+    }
 }

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

@@ -142,7 +142,8 @@ private BlockLoader getBlockLoaderFor(int shardId, Attribute attr, MappedFieldTy
         BlockLoader blockLoader = shardContext.blockLoader(getFieldName(attr), isUnsupported, fieldExtractPreference);
         MultiTypeEsField unionTypes = findUnionTypes(attr);
         if (unionTypes != null) {
-            String indexName = shardContext.ctx.index().getName();
+            // Use the fully qualified name `cluster:index-name` because multiple types are resolved on coordinator with the cluster prefix
+            String indexName = shardContext.ctx.getFullyQualifiedIndex().getName();
             Expression conversion = unionTypes.getConversionExpressionForIndex(indexName);
             return conversion == null
                 ? BlockLoader.CONSTANT_NULLS