Fix assertion error when using passthrough with contradicting fields

modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/150_tsdb.yml

@@ -423,6 +423,89 @@ dynamic templates - conflicting aliases:
   - match: { aggregations.filterA.tsids.buckets.0.key: "KGejYryCnrIkXYZdIF_Q8F8X2dfFIGKYisFh7t1RGGWOWgWU7C0RiFE" }
   - match: { aggregations.filterA.tsids.buckets.0.doc_count: 2 }
 
+---
+dynamic templates - conflicting aliases with top-level field:
+  - requires:
+      cluster_features: ["mapper.pass_through_priority"]
+      reason: support for priority in passthrough objects
+  - do:
+      allowed_warnings:
+        - "index template [my-dynamic-template] has index patterns [otel] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-dynamic-template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: my-dynamic-template
+        body:
+          index_patterns: [otel]
+          data_stream: {}
+          template:
+            settings:
+              index:
+                number_of_shards: 1
+                mode: time_series
+                time_series:
+                  start_time: 2023-08-31T13:03:08.138Z
+
+            mappings:
+              properties:
+                body:
+                  type: match_only_text
+                attributes:
+                  type: passthrough
+                  dynamic: true
+                  time_series_dimension: true
+                  priority: 1
+                scope:
+                  properties:
+                    attributes:
+                      type: passthrough
+                      dynamic: true
+                      time_series_dimension: true
+                      priority: 2
+                resource:
+                  properties:
+                    attributes:
+                      type: passthrough
+                      dynamic: true
+                      time_series_dimension: true
+                      priority: 3
+                metrics:
+                  type: passthrough
+                  dynamic: true
+                  priority: 0
+              dynamic_templates:
+                - counter_metric:
+                    mapping:
+                      type: integer
+                      time_series_metric: counter
+                      ignore_malformed: true
+                - strings_as_keyword:
+                    mapping:
+                      type: keyword
+                      ignore_above: 1024
+                      match_mapping_type: string
+                      path_match: "*attributes.*"
+
+  - do:
+      bulk:
+        index: otel
+        refresh: true
+        body:
+          - '{ "create": { "dynamic_templates": { "metrics.data": "counter_metric" } } }'
+          - '{ "@timestamp": "2023-09-01T13:03:08.138Z", "metrics": {"data": "10"}, "body": "top-level", "attributes": {"body": "attribute"}, "scope": {"attributes": {"body": "scope" }}, "resource": {"attributes": {"body": "resource" }}}'
+  - match: { errors: false }
+
+  - do:
+      search:
+        index: otel
+        body:
+          size: 1
+          fields: ["*"]
+
+  - match: { hits.total.value: 1 }
+  - match: { hits.hits.0.fields.body: [ top-level ] }
+  - match: { hits.hits.0.fields.attributes\.body: [ attribute ] }
+  - match: { hits.hits.0.fields.scope\.attributes\.body: [ scope ] }
+  - match: { hits.hits.0.fields.resource\.attributes\.body: [ resource ] }
+
 ---
 dynamic templates with nesting:
   - requires:

server/src/main/java/org/elasticsearch/index/mapper/FieldTypeLookup.java

@@ -106,21 +106,28 @@ final class FieldTypeLookup {
                         if (conflict.priority() > passThroughMapper.priority()) {
                             // Keep the conflicting field if it has higher priority.
                             passThroughFieldAliases.put(name, conflict);
-                            continue;
                         }
-                    } else if (fullNameToFieldType.containsKey(name)) {
-                        // There's an existing field or alias for the same field.
-                        continue;
-                    }
-                    MappedFieldType fieldType = fieldMapper.fieldType();
-                    fullNameToFieldType.put(name, fieldType);
-                    if (fieldType instanceof DynamicFieldType) {
-                        dynamicFieldTypes.put(name, (DynamicFieldType) fieldType);
                     }
                 }
             }
         }
 
+        for (Map.Entry<String, PassThroughObjectMapper> entry : passThroughFieldAliases.entrySet()) {
+            String name = entry.getKey();
+            if (fullNameToFieldType.containsKey(name)) {
+                // There's an existing field or alias for the same field.
+                continue;
+            }
+            Mapper mapper = entry.getValue().getMapper(name);
+            if (mapper instanceof FieldMapper fieldMapper) {
+                MappedFieldType fieldType = fieldMapper.fieldType();
+                fullNameToFieldType.put(name, fieldType);
+                if (fieldType instanceof DynamicFieldType) {
+                    dynamicFieldTypes.put(name, (DynamicFieldType) fieldType);
+                }
+            }
+        }
+
         for (MappedFieldType fieldType : RuntimeField.collectFieldTypes(runtimeFields).values()) {
             // this will override concrete fields with runtime fields that have the same name
             fullNameToFieldType.put(fieldType.name(), fieldType);

server/src/test/java/org/elasticsearch/index/mapper/FieldTypeLookupTests.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.test.ESTestCase;
 import org.hamcrest.Matchers;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -493,20 +494,36 @@ public void testAddRootAliasForConflictingPassThroughFields() {
         );
 
         FieldTypeLookup lookup = new FieldTypeLookup(
-            List.of(attributeField, resourceAttributeField),
+            randomizedList(attributeField, resourceAttributeField),
             List.of(),
-            List.of(attributes, resourceAttributes),
+            randomizedList(attributes, resourceAttributes),
             List.of()
         );
         assertEquals(attributeField.fieldType(), lookup.get("foo"));
     }
 
     public void testNoRootAliasForPassThroughFieldOnConflictingField() {
         MockFieldMapper attributeFoo = new MockFieldMapper("attributes.foo");
+        MockFieldMapper resourceAttributeFoo = new MockFieldMapper("resource.attributes.foo");
         MockFieldMapper foo = new MockFieldMapper("foo");
         PassThroughObjectMapper attributes = createPassThroughMapper("attributes", Map.of("foo", attributeFoo), 0);
+        PassThroughObjectMapper resourceAttributes = createPassThroughMapper("resource.attributes", Map.of("foo", resourceAttributeFoo), 1);
+
+        FieldTypeLookup lookup = new FieldTypeLookup(
+            randomizedList(foo, attributeFoo, resourceAttributeFoo),
+            List.of(),
+            randomizedList(attributes, resourceAttributes),
+            List.of()
+        );
 
-        FieldTypeLookup lookup = new FieldTypeLookup(List.of(foo, attributeFoo), List.of(), List.of(attributes), List.of());
         assertEquals(foo.fieldType(), lookup.get("foo"));
     }
+
+    @SafeVarargs
+    @SuppressWarnings("varargs")
+    static <T> List<T> randomizedList(T... values) {
+        ArrayList<T> list = new ArrayList<>(Arrays.asList(values));
+        Collections.shuffle(list, random());
+        return list;
+    }
 }