ESQL: Fix wrong marking of a field as unmapped when indices shared the same mapping (#133298)

A follow up fixing https://github.com/elastic/elasticsearch/pull/119886/files/e32a4463e8a4015c7be17cb984a80e79ed1e45c5#r2287545802.

The original issue, as mentioned in the comment, was triggered by multiple indices sharing the same mapping. This PR fixes it by reading the FieldCapabilitiesResponse directly instead.

Author: GalLalouche

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/IndexResolver.java

@@ -92,13 +92,15 @@ public void resolveAsMergedMapping(
 
     // public for testing only
     public static IndexResolution mergedMappings(String indexPattern, FieldCapabilitiesResponse fieldCapsResponse) {
-        var numberOfIndices = fieldCapsResponse.getIndexResponses().size();
         assert ThreadPool.assertCurrentThreadPool(ThreadPool.Names.SEARCH_COORDINATION); // too expensive to run this on a transport worker
+        var numberOfIndices = fieldCapsResponse.getIndexResponses().size();
         if (fieldCapsResponse.getIndexResponses().isEmpty()) {
             return IndexResolution.notFound(indexPattern);
         }
 
-        Map<String, List<IndexFieldCapabilities>> fieldsCaps = collectFieldCaps(fieldCapsResponse);
+        var collectedFieldCaps = collectFieldCaps(fieldCapsResponse);
+        Map<String, IndexFieldCapabilitiesWithSourceHash> fieldsCaps = collectedFieldCaps.fieldsCaps;
+        Map<String, Integer> indexMappingHashDuplicates = collectedFieldCaps.indexMappingHashDuplicates;
 
         // Build hierarchical fields - it's easier to do it in sorted order so the object fields come first.
         // TODO flattened is simpler - could we get away with that?
@@ -130,7 +132,8 @@ public static IndexResolution mergedMappings(String indexPattern, FieldCapabilit
             }
             // TODO we're careful to make isAlias match IndexResolver - but do we use it?
 
-            List<IndexFieldCapabilities> fcs = fieldsCaps.get(fullName);
+            var fieldCap = fieldsCaps.get(fullName);
+            List<IndexFieldCapabilities> fcs = fieldCap.fieldCapabilities;
             EsField field = firstUnsupportedParent == null
                 ? createField(fieldCapsResponse, name, fullName, fcs, isAlias)
                 : new UnsupportedEsField(
@@ -140,8 +143,7 @@ public static IndexResolution mergedMappings(String indexPattern, FieldCapabilit
                     new HashMap<>()
                 );
             fields.put(name, field);
-
-            var isPartiallyUnmapped = fcs.size() < numberOfIndices;
+            var isPartiallyUnmapped = fcs.size() + indexMappingHashDuplicates.getOrDefault(fieldCap.indexMappingHash, 0) < numberOfIndices;
             if (isPartiallyUnmapped) {
                 partiallyUnmappedFields.add(fullName);
             }
@@ -162,23 +164,46 @@ public static IndexResolution mergedMappings(String indexPattern, FieldCapabilit
         return IndexResolution.valid(index, concreteIndices.keySet(), failures);
     }
 
-    private static Map<String, List<IndexFieldCapabilities>> collectFieldCaps(FieldCapabilitiesResponse fieldCapsResponse) {
-        Set<String> seenHashes = new HashSet<>();
-        Map<String, List<IndexFieldCapabilities>> fieldsCaps = new HashMap<>();
+    private record IndexFieldCapabilitiesWithSourceHash(List<IndexFieldCapabilities> fieldCapabilities, String indexMappingHash) {}
+
+    private record CollectedFieldCaps(
+        Map<String, IndexFieldCapabilitiesWithSourceHash> fieldsCaps,
+        // The map won't contain entries without duplicates, i.e., it's number of occurrences - 1.
+        Map<String, Integer> indexMappingHashDuplicates
+    ) {}
+
+    private static CollectedFieldCaps collectFieldCaps(FieldCapabilitiesResponse fieldCapsResponse) {
+        Map<String, Integer> indexMappingHashToDuplicateCount = new HashMap<>();
+        Map<String, IndexFieldCapabilitiesWithSourceHash> fieldsCaps = new HashMap<>();
+
         for (FieldCapabilitiesIndexResponse response : fieldCapsResponse.getIndexResponses()) {
-            if (seenHashes.add(response.getIndexMappingHash()) == false) {
+            if (indexMappingHashToDuplicateCount.compute(response.getIndexMappingHash(), (k, v) -> v == null ? 1 : v + 1) > 1) {
                 continue;
             }
             for (IndexFieldCapabilities fc : response.get().values()) {
                 if (fc.isMetadatafield()) {
                     // ESQL builds the metadata fields if they are asked for without using the resolution.
                     continue;
                 }
-                List<IndexFieldCapabilities> all = fieldsCaps.computeIfAbsent(fc.name(), (_key) -> new ArrayList<>());
+                List<IndexFieldCapabilities> all = fieldsCaps.computeIfAbsent(
+                    fc.name(),
+                    (_key) -> new IndexFieldCapabilitiesWithSourceHash(new ArrayList<>(), response.getIndexMappingHash())
+                ).fieldCapabilities;
                 all.add(fc);
             }
         }
-        return fieldsCaps;
+
+        var iterator = indexMappingHashToDuplicateCount.entrySet().iterator();
+        while (iterator.hasNext()) {
+            var next = iterator.next();
+            if (next.getValue() <= 1) {
+                iterator.remove();
+            } else {
+                next.setValue(next.getValue() - 1);
+            }
+        }
+
+        return new CollectedFieldCaps(fieldsCaps, indexMappingHashToDuplicateCount);
     }
 
     private static EsField createField(

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/AnalyzerTests.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.fieldcaps.FieldCapabilitiesResponse;
 import org.elasticsearch.action.fieldcaps.IndexFieldCapabilities;
 import org.elasticsearch.action.fieldcaps.IndexFieldCapabilitiesBuilder;
+import org.elasticsearch.common.hash.MessageDigests;
 import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.IndexMode;
@@ -89,6 +90,7 @@
 import org.elasticsearch.xpack.esql.session.IndexResolver;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.time.Period;
 import java.util.ArrayList;
 import java.util.List;
@@ -3017,6 +3019,27 @@ public void testResolveInsist_multiIndexFieldPartiallyExistsWithMultiTypesNoKeyw
         assertThat(attr.unresolvedMessage(), is(expected));
     }
 
+    public void testResolveInsist_multiIndexSameMapping_fieldIsMapped() {
+        assumeTrue("Requires UNMAPPED FIELDS", EsqlCapabilities.Cap.UNMAPPED_FIELDS.isEnabled());
+
+        IndexResolution resolution = IndexResolver.mergedMappings(
+            "foo, bar",
+            new FieldCapabilitiesResponse(
+                List.of(
+                    fieldCapabilitiesIndexResponse("foo", messageResponseMap("long")),
+                    fieldCapabilitiesIndexResponse("bar", messageResponseMap("long"))
+                ),
+                List.of()
+            )
+        );
+        var plan = analyze("FROM foo, bar | INSIST_🐔 message", analyzer(resolution, TEST_VERIFIER));
+        var limit = as(plan, Limit.class);
+        var insist = as(limit.child(), Insist.class);
+        var attribute = (FieldAttribute) EsqlTestUtils.singleValue(insist.output());
+        assertThat(attribute.name(), is("message"));
+        assertThat(attribute.dataType(), is(DataType.LONG));
+    }
+
     public void testResolveInsist_multiIndexFieldPartiallyExistsWithMultiTypesWithKeyword_createsAnInvalidMappedField() {
         assumeTrue("Requires UNMAPPED FIELDS", EsqlCapabilities.Cap.UNMAPPED_FIELDS.isEnabled());
 
@@ -3472,7 +3495,11 @@ private static FieldCapabilitiesIndexResponse fieldCapabilitiesIndexResponse(
         String indexName,
         Map<String, IndexFieldCapabilities> fields
     ) {
-        return new FieldCapabilitiesIndexResponse(indexName, indexName, fields, false, IndexMode.STANDARD);
+        String indexMappingHash = new String(
+            MessageDigests.sha256().digest(fields.toString().getBytes(StandardCharsets.UTF_8)),
+            StandardCharsets.UTF_8
+        );
+        return new FieldCapabilitiesIndexResponse(indexName, indexMappingHash, fields, false, IndexMode.STANDARD);
     }
 
     private static Map<String, IndexFieldCapabilities> messageResponseMap(String date) {