Apply auto-flattening to subobjects: auto (#112092)

* Introduce mode `subobjects=auto` for objects

* Update docs/changelog/110524.yaml

* compilation error

* tests and fixes

* refactor

* spotless

* more tests

* fix nested objects

* fix test

* update fetch test

* add QA coverage

* update tests

* update tests

* update tests

* Apply auto-flattening to `subobjects: auto`

* Update docs/changelog/112092.yaml

* sync

* dont flatten subobjects auto

* refine test

* fix path for nested flattened objects and dynamic

* document `subobjects: auto`

* Apply suggestions from code review

Co-authored-by: Felix Barnsteiner <felixbarny@users.noreply.github.com>

* comment updates

* restore indentation in comment

* update comment

* update comment

* update comment

* update comment

* rename isFlattenable

* add test for dynamic template

* fix copy_to and noop dynamic updates

* tests

* update comment

* fix tests

* update cluster feature in yaml test

* address comments

---------

Co-authored-by: Felix Barnsteiner <felixbarny@users.noreply.github.com>

docs/changelog/112092.yaml

@@ -0,0 +1,5 @@
+pr: 112092
+summary: "Apply auto-flattening to `subobjects: auto`"
+area: Mapping
+type: enhancement
+issues: []

docs/reference/mapping/params/subobjects.asciidoc

@@ -10,7 +10,7 @@ where for instance a field `metrics.time` holds a value too, which is common whe
 A document holding a value for both `metrics.time.max` and `metrics.time` gets rejected given that `time`
 would need to be a leaf field to hold a value as well as an object to hold the `max` sub-field.
 
-The `subobjects` setting, which can be applied only to the top-level mapping definition and
+The `subobjects: false` setting, which can be applied only to the top-level mapping definition and
 to <<object,`object`>> fields, disables the ability for an object to hold further subobjects and makes it possible
 to store documents where field names contain dots and share common prefixes. From the example above, if the object
 container `metrics` has `subobjects` set to `false`, it can hold values for both `time` and `time.max` directly
@@ -109,26 +109,138 @@ PUT my-index-000001/_doc/metric_1
 <1> The entire mapping is configured to not support objects.
 <2> The document does not support objects
 
-The `subobjects` setting for existing fields and the top-level mapping definition cannot be updated.
-
-==== Auto-flattening object mappings
-
-It is generally recommended to define the properties of an object that is configured with `subobjects: false` with dotted field names
-(as shown in the first example).
-However, it is also possible to define these properties as sub-objects in the mappings.
-In that case, the mapping will be automatically flattened before it is stored.
-This makes it easier to re-use existing mappings without having to re-write them.
-
-Note that auto-flattening will not work when certain <<mapping-params, mapping parameters>> are set
-on object mappings that are defined under an object configured with `subobjects: false`:
-
-* The <<enabled, `enabled`>> mapping parameter must not be `false`.
-* The <<dynamic, `dynamic`>> mapping parameter must not contradict the implicit or explicit value of the parent. For example, when `dynamic` is set to `false` in the root of the mapping, object mappers that set `dynamic` to `true` can't be auto-flattened.
-* The <<subobjects, `subobjects`>> mapping parameter must not be set to `true` explicitly.
+Setting `subobjects: false` disallows the definition of <<object,`object`>> and <<nested,`nested`>> sub-fields, which
+can be too restrictive in cases where it's desirable to have <<nested,`nested`>> objects or sub-objects with specific
+behavior (e.g. with `enabled:false`). In this case, it's possible to set `subobjects: auto`, which
+<<auto-flattening, auto-flattens>> whenever possible and falls back to creating an object mapper otherwise (instead of
+rejecting the mapping as `subobjects: false` does). For instance:
 
 [source,console]
 --------------------------------------------------
 PUT my-index-000002
+{
+  "mappings": {
+    "properties": {
+      "metrics": {
+        "type":  "object",
+        "subobjects": "auto", <1>
+        "properties": {
+          "inner": {
+            "type": "object",
+            "enabled": false
+          },
+          "nested": {
+            "type": "nested"
+          }
+        }
+      }
+    }
+  }
+}
+
+PUT my-index-000002/_doc/metric_1
+{
+  "metrics.time" : 100, <2>
+  "metrics.time.min" : 10,
+  "metrics.time.max" : 900
+}
+
+PUT my-index-000002/_doc/metric_2
+{
+  "metrics" : {  <3>
+    "time" : 100,
+    "time.min" : 10,
+    "time.max" : 900,
+    "inner": {
+      "foo": "bar",
+      "path.to.some.field": "baz"
+    },
+    "nested": [
+      { "id": 10 },
+      { "id": 1 }
+    ]
+  }
+}
+
+GET my-index-000002/_mapping
+--------------------------------------------------
+
+[source,console-result]
+--------------------------------------------------
+{
+  "my-index-000002" : {
+    "mappings" : {
+      "properties" : {
+        "metrics" : {
+          "subobjects" : auto,
+          "properties" : {
+            "inner": {  <4>
+              "type": "object",
+              "enabled": false
+            },
+            "nested": {
+              "type": "nested",
+              "properties" : {
+                "id" : {
+                  "type" : "long"
+                }
+              }
+            },
+            "time" : {
+              "type" : "long"
+            },
+            "time.min" : {
+              "type" : "long"
+            },
+            "time.max" : {
+              "type" : "long"
+            }
+          }
+        }
+      }
+    }
+  }
+}
+--------------------------------------------------
+
+<1> The `metrics` field can only hold statically defined objects, namely `inner` and `nested`.
+<2> Sample document holding flat paths
+<3> Sample document holding an object (configured with sub-objects) and its leaf sub-fields
+<4> The resulting mapping where dots in field names (`time.min`, `time_max`), as well as the
+statically-defined sub-objects `inner` and `nested`, were preserved
+
+The `subobjects` setting for existing fields and the top-level mapping definition cannot be updated.
+
+[[auto-flattening]]
+==== Auto-flattening object mappings
+
+It is generally recommended to define the properties of an object that is configured with `subobjects: false` or
+`subobjects: auto` with dotted field names (as shown in the first example). However, it is also possible to define
+these properties as sub-objects in the mappings. In that case, the mapping will be automatically flattened before
+it is stored. This makes it easier to re-use existing mappings without having to re-write them.
+
+Note that auto-flattening does not apply if any of the following <<mapping-params, mapping parameters>> are set
+on object mappings that are defined under an object configured with `subobjects: false` or `subobjects: auto`:
+
+* The <<enabled, `enabled`>> mapping parameter is `false`.
+* The <<dynamic, `dynamic`>> mapping parameter contradicts the implicit or explicit value of the parent.
+For example, when `dynamic` is set to `false` in the root of the mapping, object mappers that set `dynamic` to `true`
+can't be auto-flattened.
+* The <<subobjects, `subobjects`>> mapping parameter is set to `auto` or `true` explicitly.
+
+If such a sub-object is detected, the behavior depends on the `subobjects` value:
+
+* `subobjects: false` is not compatible, so a mapping error is returned during mapping construction.
+* `subobjects: auto` reverts to adding the object to the mapping, bypassing auto-flattening for it. Still, any
+intermediate objects will be auto-flattened if applicable (i.e. the object name gets directly attached under the parent
+object with `subobjects: auto`). Auto-flattening can be applied within sub-objects, if they are configured with
+`subobjects: auto` too.
+
+Auto-flattening example with `subobjects: false`:
+
+[source,console]
+--------------------------------------------------
+PUT my-index-000003
 {
   "mappings": {
     "properties": {
@@ -147,13 +259,13 @@ PUT my-index-000002
     }
   }
 }
-GET my-index-000002/_mapping
+GET my-index-000003/_mapping
 --------------------------------------------------
 
 [source,console-result]
 --------------------------------------------------
 {
-  "my-index-000002" : {
+  "my-index-000003" : {
     "mappings" : {
       "properties" : {
         "metrics" : {
@@ -175,5 +287,85 @@ GET my-index-000002/_mapping
 
 <1> The metrics object can contain further object mappings that will be auto-flattened.
  Object mappings at this level must not set certain mapping parameters as explained above.
-<2> This field will be auto-flattened to `"time.min"` before the mapping is stored.
-<3> The auto-flattened `"time.min"` field can be inspected by looking at the index mapping.
+<2> This field will be auto-flattened to `time.min` before the mapping is stored.
+<3> The auto-flattened `time.min` field can be inspected by looking at the index mapping.
+
+Auto-flattening example with `subobjects: auto`:
+
+[source,console]
+--------------------------------------------------
+PUT my-index-000004
+{
+  "mappings": {
+    "properties": {
+      "metrics": {
+        "subobjects": "auto",
+        "properties": {
+          "time": {
+            "type": "object", <1>
+            "properties": {
+              "min": { "type": "long" } <2>
+            }
+          },
+          "to": {
+            "type": "object",
+            "properties": {
+              "inner_metrics": {  <3>
+                "type": "object",
+                "subobjects": "auto",
+                "properties": {
+                  "time": {
+                    "type": "object",
+                    "properties": {
+                      "max": { "type": "long" } <4>
+                    }
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+}
+GET my-index-000004/_mapping
+--------------------------------------------------
+
+[source,console-result]
+--------------------------------------------------
+{
+  "my-index-000004" : {
+    "mappings" : {
+      "properties" : {
+        "metrics" : {
+          "subobjects" : "auto",
+          "properties" : {
+            "time.min" : { <5>
+              "type" : "long"
+            },
+            "to.inner_metrics" : { <6>
+              "subobjects" : "auto",
+              "properties" : {
+                "time.max" : { <7>
+                  "type" : "long"
+                }
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+}
+--------------------------------------------------
+
+<1> The metrics object can contain further object mappings that may be auto-flattened, depending on their mapping
+parameters as explained above.
+<2> This field will be auto-flattened to `time.min` before the mapping is stored.
+<3> This object has param `subobjects: auto` so it can't be auto-flattened. Its parent does qualify for auto-flattening,
+so it becomes `to.inner_metrics` before the mapping is stored.
+<4> This field will be auto-flattened to `time.max` before the mapping is stored.
+<5> The auto-flattened `time.min` field can be inspected by looking at the index mapping.
+<6> The inner object `to.inner_metrics` can be inspected by looking at the index mapping.
+<7> The auto-flattened `time.max` field can be inspected by looking at the index mapping.

modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/logsdb/qa/DataGenerationHelper.java

@@ -35,12 +35,7 @@ class DataGenerationHelper {
     private final DataGenerator dataGenerator;
 
     DataGenerationHelper() {
-        // TODO enable subobjects: auto
-        // It is disabled because it currently does not have auto flattening and that results in asserts being triggered when using copy_to.
-        this.subobjects = ESTestCase.randomValueOtherThan(
-            ObjectMapper.Subobjects.AUTO,
-            () -> ESTestCase.randomFrom(ObjectMapper.Subobjects.values())
-        );
+        this.subobjects = ESTestCase.randomFrom(ObjectMapper.Subobjects.values());
         this.keepArraySource = ESTestCase.randomBoolean();
 
         var specificationBuilder = DataGeneratorSpecification.builder().withFullyDynamicMapping(ESTestCase.randomBoolean());

modules/dot-prefix-validation/build.gradle

@@ -27,3 +27,13 @@ tasks.named('yamlRestTest') {
 tasks.named('yamlRestCompatTest') {
   usesDefaultDistribution()
 }
+
+tasks.named("yamlRestCompatTestTransform").configure(
+  { task ->
+    task.skipTest("tsdb/140_routing_path/multi-value routing path field", "Multi-value routing paths are allowed now. See #112645")
+    task.skipTest(
+      "dot_prefix/10_basic/Deprecated index template with a dot prefix index pattern",
+      "Tentantively disabled until #112092 gets backported to 8.x"
+    )
+  }
+)

rest-api-spec/build.gradle

@@ -54,9 +54,29 @@ tasks.named("precommit").configure {
   dependsOn 'enforceYamlTestConvention'
 }
 
-tasks.named("yamlRestCompatTestTransform").configure({ task ->
-  task.skipTest("tsdb/140_routing_path/multi-value routing path field", "Multi-value routing paths are allowed now. See #112645")
-  task.skipTest("indices.sort/10_basic/Index Sort", "warning does not exist for compatibility")
-  task.skipTest("search/330_fetch_fields/Test search rewrite", "warning does not exist for compatibility")
-  task.skipTest("search/540_ignore_above_synthetic_source/ignore_above mapping level setting on arrays", "Temporary mute while backporting to 8.x")
+tasks.named("yamlRestCompatTestTransform").configure({task ->
+    task.skipTest("tsdb/140_routing_path/multi-value routing path field", "Multi-value routing paths are allowed now. See #112645")
+    task.skipTest("indices.sort/10_basic/Index Sort", "warning does not exist for compatibility")
+    task.skipTest("search/330_fetch_fields/Test search rewrite", "warning does not exist for compatibility")
+    task.skipTest("search/540_ignore_above_synthetic_source/ignore_above mapping level setting on arrays", "Temporary mute while backporting to 8.x")
+    task.skipTest("indices.create/20_synthetic_source/subobjects auto", "Tentantively disabled until #112092 gets backported to 8.x")
+    task.skipTest(
+      "index/92_metrics_auto_subobjects/Metrics object indexing with synthetic source",
+      "Tentantively disabled until #112092 gets backported to 8.x"
+    )
+    task.skipTest(
+      "index/92_metrics_auto_subobjects/Root without subobjects with synthetic source",
+      "Tentantively disabled until #112092 gets backported to 8.x"
+    )
+    task.skipTest(
+      "indices.put_index_template/15_composition/Composable index templates that include subobjects: auto at root",
+      "Tentantively disabled until #112092 gets backported to 8.x"
+    )
+    task.skipTest(
+      "indices.put_index_template/15_composition/Composable index templates that include subobjects: auto on arbitrary field",
+      "Tentantively disabled until #112092 gets backported to 8.x"
+    )
+    task.skipTest("index/92_metrics_auto_subobjects/Metrics object indexing", "Tentantively disabled until #112092 gets backported to 8.x")
+    task.skipTest("index/92_metrics_auto_subobjects/Root with metrics", "Tentantively disabled until #112092 gets backported to 8.x")
+    task.skipTest("search/330_fetch_fields/Test with subobjects: auto", "Tentantively disabled until #112092 gets backported to 8.x")
 })

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/index/92_metrics_auto_subobjects.yml

@@ -2,7 +2,7 @@
 "Metrics object indexing":
   - requires:
       test_runner_features: [ "allowed_warnings", "allowed_warnings_regex" ]
-      cluster_features: ["mapper.subobjects_auto"]
+      cluster_features: ["mapper.subobjects_auto_fixes"]
       reason: requires supporting subobjects auto setting
 
   - do:
@@ -69,7 +69,7 @@
 "Root with metrics":
   - requires:
       test_runner_features: [ "allowed_warnings", "allowed_warnings_regex" ]
-      cluster_features: ["mapper.subobjects_auto"]
+      cluster_features: ["mapper.subobjects_auto_fixes"]
       reason: requires supporting subobjects auto setting
 
   - do:
@@ -131,7 +131,7 @@
 "Metrics object indexing with synthetic source":
   - requires:
       test_runner_features: [ "allowed_warnings", "allowed_warnings_regex" ]
-      cluster_features: ["mapper.subobjects_auto"]
+      cluster_features: ["mapper.subobjects_auto_fixes"]
       reason: added in 8.4.0
 
   - do:
@@ -201,7 +201,7 @@
 "Root without subobjects with synthetic source":
   - requires:
       test_runner_features: [ "allowed_warnings", "allowed_warnings_regex" ]
-      cluster_features: ["mapper.subobjects_auto"]
+      cluster_features: ["mapper.subobjects_auto_fixes"]
       reason: added in 8.4.0
 
   - do:

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/indices.create/20_synthetic_source.yml

@@ -887,7 +887,7 @@ doubly nested object:
 ---
 subobjects auto:
   - requires:
-      cluster_features: ["mapper.subobjects_auto"]
+      cluster_features: ["mapper.subobjects_auto_fixes"]
       reason: requires tracking ignored source and supporting subobjects auto setting
 
   - do:
@@ -924,9 +924,21 @@ subobjects auto:
                         type: keyword
               nested:
                 type: nested
-              auto_obj:
-                type: object
-                subobjects: auto
+              path:
+                properties:
+                  to:
+                    properties:
+                      auto_obj:
+                        type: object
+                        subobjects: auto
+                        properties:
+                          inner:
+                            properties:
+                              id:
+                                type: keyword
+                      id:
+                        type:
+                          integer
 
   - do:
       bulk:
@@ -934,13 +946,13 @@ subobjects auto:
         refresh: true
         body:
           - '{ "create": { } }'
-          - '{  "id": 1, "foo": 10, "foo.bar": 100, "regular": [ { "trace": { "id": "a" }, "span": { "id": "1" } }, { "trace": { "id": "b" }, "span": { "id": "1" } } ] }'
+          - '{  "id": 1, "foo": 10, "foo.bar": 100, "regular.trace.id": ["b", "a", "b"], "regular.span.id": "1" }'
           - '{ "create": { } }'
           - '{  "id": 2, "foo": 20, "foo.bar": 200, "stored": [ { "trace": { "id": "a" }, "span": { "id": "1" } }, { "trace": { "id": "b" }, "span": { "id": "1" } } ] }'
           - '{ "create": { } }'
           - '{  "id": 3, "foo": 30, "foo.bar": 300, "nested": [ { "a": 10, "b": 20 }, { "a": 100, "b": 200 } ] }'
           - '{ "create": { } }'
-          - '{  "id": 4, "auto_obj": {  "foo": 40, "foo.bar": 400 } }'
+          - '{  "id": 4, "path.to.auto_obj": {  "foo": 40, "foo.bar": 400, "inner.id": "baz" }, "path.to.id": 4000 }'
 
   - match: { errors: false }
 
@@ -952,8 +964,8 @@ subobjects auto:
   - match: { hits.hits.0._source.id: 1  }
   - match: { hits.hits.0._source.foo: 10  }
   - match: { hits.hits.0._source.foo\.bar: 100  }
-  - match: { hits.hits.0._source.regular.span.id: "1" }
-  - match: { hits.hits.0._source.regular.trace.id: [ "a", "b" ] }
+  - match: { hits.hits.0._source.regular\.span\.id: "1" }
+  - match: { hits.hits.0._source.regular\.trace\.id: [ "a", "b" ] }
   - match: { hits.hits.1._source.id: 2  }
   - match: { hits.hits.1._source.foo: 20 }
   - match: { hits.hits.1._source.foo\.bar: 200 }
@@ -969,8 +981,110 @@ subobjects auto:
   - match: { hits.hits.2._source.nested.1.a: 100 }
   - match: { hits.hits.2._source.nested.1.b: 200 }
   - match: { hits.hits.3._source.id: 4  }
-  - match: { hits.hits.3._source.auto_obj.foo: 40 }
-  - match: { hits.hits.3._source.auto_obj.foo\.bar: 400 }
+  - match: { hits.hits.3._source.path\.to\.auto_obj.foo: 40 }
+  - match: { hits.hits.3._source.path\.to\.auto_obj.foo\.bar: 400 }
+  - match: { hits.hits.3._source.path\.to\.auto_obj.inner\.id: baz }
+  - match: { hits.hits.3._source.path\.to\.id: 4000 }
+
+
+---
+subobjects auto with path flattening:
+  - requires:
+      cluster_features: ["mapper.subobjects_auto_fixes"]
+      reason: requires tracking ignored source and supporting subobjects auto setting
+
+  - do:
+      indices.create:
+        index: test
+        body:
+          mappings:
+            _source:
+              mode: synthetic
+            subobjects: auto
+            properties:
+              id:
+                type: integer
+              attributes:
+                type: object
+                subobjects: auto
+
+  - do:
+      bulk:
+        index: test
+        refresh: true
+        body:
+          - '{ "create": { } }'
+          - '{  "id": 1, "attributes": { "foo": { "bar": 10 } } }'
+          - '{ "create": { } }'
+          - '{  "id": 2, "attributes": { "foo": { "bar": 20 } } }'
+          - '{ "create": { } }'
+          - '{  "id": 3, "attributes": { "foo": { "bar": 30 } } }'
+          - '{ "create": { } }'
+          - '{  "id": 4, "attributes": { "foo": { "bar": 40 } } }'
+
+  - match: { errors: false }
+
+  - do:
+      search:
+        index: test
+        sort: id
+
+  - match: { hits.hits.0._source.id: 1  }
+  - match: { hits.hits.0._source.attributes.foo\.bar: 10  }
+  - match: { hits.hits.1._source.id: 2 }
+  - match: { hits.hits.1._source.attributes.foo\.bar: 20 }
+  - match: { hits.hits.2._source.id: 3 }
+  - match: { hits.hits.2._source.attributes.foo\.bar: 30 }
+  - match: { hits.hits.3._source.id: 4 }
+  - match: { hits.hits.3._source.attributes.foo\.bar: 40 }
+
+
+---
+subobjects auto with dynamic template:
+  - requires:
+      cluster_features: ["mapper.subobjects_auto_fixes"]
+      reason: requires tracking ignored source and supporting subobjects auto setting
+
+  - do:
+      indices.create:
+        index: test
+        body:
+          mappings:
+            _source:
+              mode: synthetic
+            subobjects: auto
+            dynamic_templates:
+              - attributes_tmpl:
+                  match: attributes
+                  mapping:
+                    type: object
+                    enabled: false
+                    subobjects: auto
+            properties:
+              id:
+                type: integer
+
+  - do:
+      bulk:
+        index: test
+        refresh: true
+        body:
+          - '{ "create": { } }'
+          - '{  "id": 1, "attributes": { "foo": 10, "path.to.bar": "val1" }, "a": 100, "a.b": 1000 }'
+
+  - match: { errors: false }
+
+  - do:
+      search:
+        index: test
+        sort: id
+
+  - match: { hits.hits.0._source.id: 1 }
+  - match: { hits.hits.0._source.attributes.foo: 10 }
+  - match: { hits.hits.0._source.attributes.path\.to\.bar: val1 }
+  - match: { hits.hits.0._source.a: 100 }
+  - match: { hits.hits.0._source.a\.b: 1000 }
+
 
 ---
 synthetic_source with copy_to:
@@ -1755,7 +1869,7 @@ synthetic_source with copy_to pointing to ambiguous field and subobjects false:
 ---
 synthetic_source with copy_to pointing to ambiguous field and subobjects auto:
   - requires:
-      cluster_features: ["mapper.source.synthetic_source_copy_to_inside_objects_fix"]
+      cluster_features: ["mapper.subobjects_auto_fixes"]
       reason: requires copy_to support in synthetic source
 
   - do:

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/indices.put_index_template/15_composition.yml

@@ -453,7 +453,7 @@
 ---
 "Composable index templates that include subobjects: auto at root":
   - requires:
-      cluster_features: ["mapper.subobjects_auto"]
+      cluster_features: ["mapper.subobjects_auto_fixes"]
       reason: "https://github.com/elastic/elasticsearch/issues/96768 fixed at 8.11.0"
       test_runner_features: "allowed_warnings"
 
@@ -504,7 +504,7 @@
 ---
 "Composable index templates that include subobjects: auto on arbitrary field":
   - requires:
-      cluster_features: ["mapper.subobjects_auto"]
+      cluster_features: ["mapper.subobjects_auto_fixes"]
       reason: "https://github.com/elastic/elasticsearch/issues/96768 fixed at 8.11.0"
       test_runner_features: "allowed_warnings"
 

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/search/330_fetch_fields.yml

@@ -1129,7 +1129,7 @@ fetch geo_point:
 ---
 "Test with subobjects: auto":
   - requires:
-      cluster_features: "mapper.subobjects_auto"
+      cluster_features: "mapper.subobjects_auto_fixes"
       reason: requires support for subobjects auto setting
 
   - do:

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

@@ -389,6 +389,14 @@ public final class DocumentParser {
             rootBuilder.addRuntimeField(runtimeField);
         }
         RootObjectMapper root = rootBuilder.build(MapperBuilderContext.root(context.mappingLookup().isSourceSynthetic(), false));
+
+        // Repeat the check, in case the dynamic mappers don't produce a mapping update.
+        // For instance, the parsed source may contain intermediate objects that get flattened,
+        // leading to an empty dynamic update.
+        if (root.mappers.isEmpty() && root.runtimeFields().isEmpty()) {
+            return null;
+        }
+
         return context.mappingLookup().getMapping().mappingUpdate(root);
     }
 
@@ -638,7 +646,7 @@ public final class DocumentParser {
     private static void doParseObject(DocumentParserContext context, String currentFieldName, Mapper objectMapper) throws IOException {
         context.path().add(currentFieldName);
         boolean withinLeafObject = context.path().isWithinLeafObject();
-        if (objectMapper instanceof ObjectMapper objMapper && objMapper.subobjects() != ObjectMapper.Subobjects.ENABLED) {
+        if (objectMapper instanceof ObjectMapper objMapper && objMapper.subobjects() == ObjectMapper.Subobjects.DISABLED) {
             context.path().setWithinLeafObject(true);
         }
         parseObjectOrField(context, objectMapper);
@@ -1012,11 +1020,15 @@ public final class DocumentParser {
         // don't create a dynamic mapping for it and don't index it.
         String fieldPath = context.path().pathAsText(fieldName);
         MappedFieldType fieldType = context.mappingLookup().getFieldType(fieldPath);
-        if (fieldType != null) {
-            // we haven't found a mapper with this name above, which means if a field type is found it is for sure a runtime field.
-            assert fieldType.hasDocValues() == false && fieldType.isAggregatable() && fieldType.isSearchable();
+
+        if (fieldType != null && fieldType.hasDocValues() == false && fieldType.isAggregatable() && fieldType.isSearchable()) {
+            // We haven't found a mapper with this name above, which means it is a runtime field.
             return noopFieldMapper(fieldPath);
         }
+        // No match or the matching field type corresponds to a mapper with flattened name (containing dots),
+        // e.g. for field 'foo.bar' under root there is no 'bar' mapper in object 'bar'.
+        // Returning null leads to creating a dynamic mapper. In the case of a mapper with flattened name,
+        // the dynamic mapper later gets deduplicated when building the dynamic update for the doc at hand.
         return null;
     }
 
@@ -1160,11 +1172,10 @@ public final class DocumentParser {
                 mappingLookup.getMapping().getRoot(),
                 ObjectMapper.Dynamic.getRootDynamic(mappingLookup)
             );
-            if (mappingLookup.getMapping().getRoot().subobjects() == ObjectMapper.Subobjects.ENABLED) {
-                this.parser = DotExpandingXContentParser.expandDots(parser, this.path);
-            } else {
-                this.parser = parser;
-            }
+            // If root supports no subobjects, there's no point in expanding dots in names to subobjects.
+            this.parser = (mappingLookup.getMapping().getRoot().subobjects() == ObjectMapper.Subobjects.DISABLED)
+                ? parser
+                : DotExpandingXContentParser.expandDots(parser, this.path, this);
             this.document = new LuceneDocument();
             this.documents.add(document);
             this.maxAllowedNumNestedDocs = indexSettings().getMappingNestedDocsLimit();

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

@@ -123,6 +123,7 @@ public abstract class DocumentParserContext {
     private Field version;
     private final SeqNoFieldMapper.SequenceIDFields seqID;
     private final Set<String> fieldsAppliedFromTemplates;
+    private final boolean supportsObjectAutoFlattening;
 
     /**
      * Fields that are copied from values of other fields via copy_to.
@@ -177,6 +178,7 @@ public abstract class DocumentParserContext {
         this.copyToFields = copyToFields;
         this.dynamicMappersSize = dynamicMapperSize;
         this.recordedSource = recordedSource;
+        this.supportsObjectAutoFlattening = checkForAutoFlatteningSupport();
     }
 
     private DocumentParserContext(ObjectMapper parent, ObjectMapper.Dynamic dynamic, DocumentParserContext in) {
@@ -204,6 +206,43 @@ public abstract class DocumentParserContext {
         );
     }
 
+    private boolean checkForAutoFlatteningSupport() {
+        if (root().subobjects() != ObjectMapper.Subobjects.ENABLED) {
+            return true;
+        }
+        for (ObjectMapper objectMapper : mappingLookup.objectMappers().values()) {
+            if (objectMapper.subobjects() != ObjectMapper.Subobjects.ENABLED) {
+                return true;
+            }
+        }
+        if (root().dynamicTemplates() != null) {
+            for (DynamicTemplate dynamicTemplate : root().dynamicTemplates()) {
+                if (findSubobjects(dynamicTemplate.getMapping())) {
+                    return true;
+                }
+            }
+        }
+        for (ObjectMapper objectMapper : dynamicObjectMappers.values()) {
+            if (objectMapper.subobjects() != ObjectMapper.Subobjects.ENABLED) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    @SuppressWarnings("unchecked")
+    private static boolean findSubobjects(Map<String, Object> mapping) {
+        for (var entry : mapping.entrySet()) {
+            if (entry.getKey().equals("subobjects") && (entry.getValue() instanceof Boolean || entry.getValue() instanceof String)) {
+                return true;
+            }
+            if (entry.getValue() instanceof Map<?, ?> && findSubobjects((Map<String, Object>) entry.getValue())) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     protected DocumentParserContext(
         MappingLookup mappingLookup,
         MappingParserContext mappingParserContext,
@@ -464,6 +503,10 @@ public abstract class DocumentParserContext {
         return copyToFields;
     }
 
+    boolean supportsObjectAutoFlattening() {
+        return supportsObjectAutoFlattening;
+    }
+
     /**
      * Add a new mapper dynamically created while parsing.
      *
@@ -599,6 +642,25 @@ public abstract class DocumentParserContext {
         return dynamicObjectMappers.get(name);
     }
 
+    ObjectMapper findObject(String fullName) {
+        // does the object mapper already exist? if so, use that
+        ObjectMapper objectMapper = mappingLookup().objectMappers().get(fullName);
+        if (objectMapper != null) {
+            return objectMapper;
+        }
+        // has the object mapper been added as a dynamic update already?
+        return getDynamicObjectMapper(fullName);
+    }
+
+    ObjectMapper.Builder findObjectBuilder(String fullName) {
+        // does the object mapper already exist? if so, use that
+        ObjectMapper objectMapper = findObject(fullName);
+        if (objectMapper != null) {
+            return objectMapper.newBuilder(indexSettings().getIndexVersionCreated());
+        }
+        return null;
+    }
+
     /**
      * Add a new runtime field dynamically created while parsing.
      * We use the same set for both new indexed and new runtime fields,
@@ -698,7 +760,7 @@ public abstract class DocumentParserContext {
      */
     public final DocumentParserContext createCopyToContext(String copyToField, LuceneDocument doc) throws IOException {
         ContentPath path = new ContentPath();
-        XContentParser parser = DotExpandingXContentParser.expandDots(new CopyToParser(copyToField, parser()), path);
+        XContentParser parser = DotExpandingXContentParser.expandDots(new CopyToParser(copyToField, parser()), path, this);
         return new Wrapper(root(), this) {
             @Override
             public ContentPath path() {

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

@@ -18,6 +18,8 @@ import org.elasticsearch.xcontent.XContentSubParser;
 
 import java.io.IOException;
 import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Deque;
 import java.util.List;
 import java.util.Map;
@@ -38,9 +40,13 @@ class DotExpandingXContentParser extends FilterXContentParserWrapper {
 
         private final ContentPath contentPath;
         final Deque<XContentParser> parsers = new ArrayDeque<>();
+        final DocumentParserContext context;
+        boolean supportsObjectAutoFlattening;
 
-        WrappingParser(XContentParser in, ContentPath contentPath) throws IOException {
+        WrappingParser(XContentParser in, ContentPath contentPath, DocumentParserContext context) throws IOException {
             this.contentPath = contentPath;
+            this.context = context;
+            this.supportsObjectAutoFlattening = (context != null && context.supportsObjectAutoFlattening());
             parsers.push(in);
             if (in.currentToken() == Token.FIELD_NAME) {
                 expandDots(in);
@@ -107,7 +113,7 @@ class DotExpandingXContentParser extends FilterXContentParserWrapper {
             if (resultSize == 0) {
                 throw new IllegalArgumentException("field name cannot contain only dots");
             }
-            final String[] subpaths;
+            String[] subpaths;
             if (resultSize == list.length) {
                 for (String part : list) {
                     // check if the field name contains only whitespace
@@ -126,6 +132,9 @@ class DotExpandingXContentParser extends FilterXContentParserWrapper {
                 }
                 subpaths = extractAndValidateResults(field, list, resultSize);
             }
+            if (supportsObjectAutoFlattening && subpaths.length > 1) {
+                subpaths = maybeFlattenPaths(Arrays.asList(subpaths), context, contentPath).toArray(String[]::new);
+            }
             pushSubParser(delegate, subpaths);
         }
 
@@ -235,11 +244,13 @@ class DotExpandingXContentParser extends FilterXContentParserWrapper {
 
     /**
      * Wraps an XContentParser such that it re-interprets dots in field names as an object structure
-     * @param in    the parser to wrap
-     * @return  the wrapped XContentParser
+     * @param in            the parser to wrap
+     * @param contentPath   the starting path to expand, can be empty
+     * @param context       provides mapping context to check for objects supporting sub-object auto-flattening
+     * @return              the wrapped XContentParser
      */
-    static XContentParser expandDots(XContentParser in, ContentPath contentPath) throws IOException {
-        return new WrappingParser(in, contentPath);
+    static XContentParser expandDots(XContentParser in, ContentPath contentPath, DocumentParserContext context) throws IOException {
+        return new WrappingParser(in, contentPath, context);
     }
 
     private enum State {
@@ -410,4 +421,49 @@ class DotExpandingXContentParser extends FilterXContentParserWrapper {
             return null;
         }
     }
+
+    static List<String> maybeFlattenPaths(List<String> subpaths, DocumentParserContext context, ContentPath contentPath) {
+        String prefixWithDots = contentPath.pathAsText("");
+        ObjectMapper parent = contentPath.length() == 0
+            ? context.root()
+            : context.findObject(prefixWithDots.substring(0, prefixWithDots.length() - 1));
+        List<String> result = new ArrayList<>(subpaths.size());
+        for (int i = 0; i < subpaths.size(); i++) {
+            String fullPath = prefixWithDots + String.join(".", subpaths.subList(0, i));
+            if (i > 0) {
+                parent = context.findObject(fullPath);
+            }
+            boolean match = false;
+            StringBuilder path = new StringBuilder(subpaths.get(i));
+            if (parent == null) {
+                // We get here for dynamic objects, which always get parsed with subobjects and may get flattened later.
+                match = true;
+            } else if (parent.subobjects() == ObjectMapper.Subobjects.ENABLED) {
+                match = true;
+            } else if (parent.subobjects() == ObjectMapper.Subobjects.AUTO) {
+                // Check if there's any subobject in the remaining path.
+                for (int j = i; j < subpaths.size() - 1; j++) {
+                    if (j > i) {
+                        path.append(".").append(subpaths.get(j));
+                    }
+                    Mapper mapper = parent.mappers.get(path.toString());
+                    if (mapper instanceof ObjectMapper objectMapper
+                        && (ObjectMapper.isFlatteningCandidate(objectMapper.subobjects, objectMapper)
+                            || objectMapper.checkFlattenable(null).isPresent())) {
+                        i = j;
+                        match = true;
+                        break;
+                    }
+                }
+            }
+            if (match) {
+                result.add(path.toString());
+            } else {
+                // We only get here if parent has subobjects set to false, or set to auto with no non-flattenable object in the sub-path.
+                result.add(String.join(".", subpaths.subList(i, subpaths.size())));
+                return result;
+            }
+        }
+        return result;
+    }
 }

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

@@ -21,6 +21,7 @@ import org.elasticsearch.xcontent.XContentParser;
 import java.io.IOException;
 import java.time.DateTimeException;
 import java.util.Map;
+import java.util.Optional;
 
 /**
  * Encapsulates the logic for dynamically creating fields as part of document parsing.
@@ -162,7 +163,9 @@ final class DynamicFieldsBuilder {
         Mapper mapper = createObjectMapperFromTemplate(context, name);
         return mapper != null
             ? mapper
-            : new ObjectMapper.Builder(name, context.parent().subobjects).enabled(ObjectMapper.Defaults.ENABLED)
+            // Dynamic objects are configured with subobject support, otherwise they can't get auto-flattened
+            // even if they otherwise qualify.
+            : new ObjectMapper.Builder(name, Optional.empty()).enabled(ObjectMapper.Defaults.ENABLED)
                 .build(context.createDynamicMapperBuilderContext());
     }
 

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

@@ -36,6 +36,7 @@ public class MapperFeatures implements FeatureSpecification {
             NodeMappingStats.SEGMENT_LEVEL_FIELDS_STATS,
             BooleanFieldMapper.BOOLEAN_DIMENSION,
             ObjectMapper.SUBOBJECTS_AUTO,
+            ObjectMapper.SUBOBJECTS_AUTO_FIXES,
             KeywordFieldMapper.KEYWORD_NORMALIZER_SYNTHETIC_SOURCE,
             SourceFieldMapper.SYNTHETIC_SOURCE_STORED_FIELDS_ADVANCE_FIX,
             Mapper.SYNTHETIC_SOURCE_KEEP_FEATURE,

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

@@ -45,6 +45,7 @@ public class ObjectMapper extends Mapper {
     public static final String CONTENT_TYPE = "object";
     static final String STORE_ARRAY_SOURCE_PARAM = "store_array_source";
     static final NodeFeature SUBOBJECTS_AUTO = new NodeFeature("mapper.subobjects_auto");
+    static final NodeFeature SUBOBJECTS_AUTO_FIXES = new NodeFeature("mapper.subobjects_auto_fixes");
 
     /**
      * Enhances the previously boolean option for subobjects support with an intermediate mode `auto` that uses
@@ -176,42 +177,84 @@ public class ObjectMapper extends Mapper {
             // If the mapper to add has no dots, or the current object mapper has subobjects set to false,
             // we just add it as it is for sure a leaf mapper
             if (name.contains(".") == false || (subobjects.isPresent() && (subobjects.get() == Subobjects.DISABLED))) {
-                add(name, mapper);
-            } else {
-                // We strip off the first object path of the mapper name, load or create
-                // the relevant object mapper, and then recurse down into it, passing the remainder
-                // of the mapper name. So for a mapper 'foo.bar.baz', we locate 'foo' and then
-                // call addDynamic on it with the name 'bar.baz', and next call addDynamic on 'bar' with the name 'baz'.
-                int firstDotIndex = name.indexOf('.');
-                String immediateChild = name.substring(0, firstDotIndex);
-                String immediateChildFullName = prefix == null ? immediateChild : prefix + "." + immediateChild;
-                Builder parentBuilder = findObjectBuilder(immediateChildFullName, context);
-                if (parentBuilder != null) {
-                    parentBuilder.addDynamic(name.substring(firstDotIndex + 1), immediateChildFullName, mapper, context);
-                    add(parentBuilder);
-                } else if (subobjects.isPresent() && subobjects.get() == Subobjects.AUTO) {
-                    // No matching parent object was found, the mapper is added as a leaf - similar to subobjects false.
-                    add(name, mapper);
-                } else {
-                    // Expected to find a matching parent object but got null.
-                    throw new IllegalStateException("Missing intermediate object " + immediateChildFullName);
+                if (mapper instanceof ObjectMapper objectMapper
+                    && isFlatteningCandidate(subobjects, objectMapper)
+                    && objectMapper.checkFlattenable(null).isEmpty()) {
+                    // Subobjects auto and false don't allow adding subobjects dynamically.
+                    return;
                 }
+                add(name, mapper);
+                return;
             }
-        }
+            if (subobjects.isPresent() && subobjects.get() == Subobjects.AUTO) {
+                // Check if there's an existing field with the sanme, to avoid no-op dynamic updates.
+                ObjectMapper objectMapper = (prefix == null) ? context.root() : context.mappingLookup().objectMappers().get(prefix);
+                if (objectMapper != null && objectMapper.mappers.containsKey(name)) {
+                    return;
+                }
 
-        private static Builder findObjectBuilder(String fullName, DocumentParserContext context) {
-            // does the object mapper already exist? if so, use that
-            ObjectMapper objectMapper = context.mappingLookup().objectMappers().get(fullName);
-            if (objectMapper != null) {
-                return objectMapper.newBuilder(context.indexSettings().getIndexVersionCreated());
+                // Check for parent objects. Due to auto-flattening, names with dots are allowed so we need to check for all possible
+                // object names. For instance, for mapper 'foo.bar.baz.bad', we have the following options:
+                // -> object 'foo' found => call addDynamic on 'bar.baz.bad'
+                // ---> object 'bar' found => call addDynamic on 'baz.bad'
+                // -----> object 'baz' found => add field 'bad' to it
+                // -----> no match found => add field 'baz.bad' to 'bar'
+                // ---> object 'bar.baz' found => add field 'bad' to it
+                // ---> no match found => add field 'bar.baz.bad' to 'foo'
+                // -> object 'foo.bar' found => call addDynamic on 'baz.bad'
+                // ---> object 'baz' found => add field 'bad' to it
+                // ---> no match found=> add field 'baz.bad' to 'foo.bar'
+                // -> object 'foo.bar.baz' found => add field 'bad' to it
+                // -> no match found => add field 'foo.bar.baz.bad' to parent
+                String fullPathToMapper = name.substring(0, name.lastIndexOf(mapper.leafName()));
+                String[] fullPathTokens = fullPathToMapper.split("\\.");
+                StringBuilder candidateObject = new StringBuilder();
+                String candidateObjectPrefix = prefix == null ? "" : prefix + ".";
+                for (int i = 0; i < fullPathTokens.length; i++) {
+                    if (candidateObject.isEmpty() == false) {
+                        candidateObject.append(".");
+                    }
+                    candidateObject.append(fullPathTokens[i]);
+                    String candidateFullObject = candidateObjectPrefix.isEmpty()
+                        ? candidateObject.toString()
+                        : candidateObjectPrefix + candidateObject.toString();
+                    ObjectMapper parent = context.findObject(candidateFullObject);
+                    if (parent != null) {
+                        var parentBuilder = parent.newBuilder(context.indexSettings().getIndexVersionCreated());
+                        parentBuilder.addDynamic(name.substring(candidateObject.length() + 1), candidateFullObject, mapper, context);
+                        if (parentBuilder.mappersBuilders.isEmpty() == false) {
+                            add(parentBuilder);
+                        }
+                        return;
+                    }
+                }
+
+                // No matching parent object was found, the mapper is added as a leaf - similar to subobjects false.
+                // This only applies to field mappers, as subobjects get auto-flattened.
+                if (mapper instanceof FieldMapper fieldMapper) {
+                    FieldMapper.Builder fieldBuilder = fieldMapper.getMergeBuilder();
+                    fieldBuilder.setLeafName(name);  // Update to reflect the current, possibly flattened name.
+                    add(fieldBuilder);
+                }
+                return;
             }
-            // has the object mapper been added as a dynamic update already?
-            objectMapper = context.getDynamicObjectMapper(fullName);
-            if (objectMapper != null) {
-                return objectMapper.newBuilder(context.indexSettings().getIndexVersionCreated());
+
+            // We strip off the first object path of the mapper name, load or create
+            // the relevant object mapper, and then recurse down into it, passing the remainder
+            // of the mapper name. So for a mapper 'foo.bar.baz', we locate 'foo' and then
+            // call addDynamic on it with the name 'bar.baz', and next call addDynamic on 'bar' with the name 'baz'.
+            int firstDotIndex = name.indexOf('.');
+            String immediateChild = name.substring(0, firstDotIndex);
+            String immediateChildFullName = prefix == null ? immediateChild : prefix + "." + immediateChild;
+            Builder parentBuilder = context.findObjectBuilder(immediateChildFullName);
+            if (parentBuilder != null) {
+                parentBuilder.addDynamic(name.substring(firstDotIndex + 1), immediateChildFullName, mapper, context);
+                add(parentBuilder);
+            } else {
+                // Expected to find a matching parent object but got null.
+                throw new IllegalStateException("Missing intermediate object " + immediateChildFullName);
             }
-            // no object mapper found
-            return null;
+
         }
 
         protected final Map<String, Mapper> buildMappers(MapperBuilderContext mapperBuilderContext) {
@@ -227,9 +270,10 @@ public class ObjectMapper extends Mapper {
                     // mix of object notation and dot notation.
                     mapper = existing.merge(mapper, MapperMergeContext.from(mapperBuilderContext, Long.MAX_VALUE));
                 }
-                if (subobjects.isPresent() && subobjects.get() == Subobjects.DISABLED && mapper instanceof ObjectMapper objectMapper) {
-                    // We're parsing a mapping that has set `subobjects: false` but has defined sub-objects
-                    objectMapper.asFlattenedFieldMappers(mapperBuilderContext).forEach(m -> mappers.put(m.leafName(), m));
+                if (mapper instanceof ObjectMapper objectMapper && isFlatteningCandidate(subobjects, objectMapper)) {
+                    // We're parsing a mapping that has defined sub-objects, may need to flatten them.
+                    objectMapper.asFlattenedFieldMappers(mapperBuilderContext, throwOnFlattenableError(subobjects))
+                        .forEach(m -> mappers.put(m.leafName(), m));
                 } else {
                     mappers.put(mapper.leafName(), mapper);
                 }
@@ -624,12 +668,11 @@ public class ObjectMapper extends Mapper {
             Optional<Subobjects> subobjects
         ) {
             Map<String, Mapper> mergedMappers = new HashMap<>();
+            var context = objectMergeContext.getMapperBuilderContext();
             for (Mapper childOfExistingMapper : existing.mappers.values()) {
-                if (subobjects.isPresent()
-                    && subobjects.get() == Subobjects.DISABLED
-                    && childOfExistingMapper instanceof ObjectMapper objectMapper) {
-                    // An existing mapping with sub-objects is merged with a mapping that has set `subobjects: false`
-                    objectMapper.asFlattenedFieldMappers(objectMergeContext.getMapperBuilderContext())
+                if (childOfExistingMapper instanceof ObjectMapper objectMapper && isFlatteningCandidate(subobjects, objectMapper)) {
+                    // An existing mapping with sub-objects is merged with a mapping that has `subobjects` set to false or auto.
+                    objectMapper.asFlattenedFieldMappers(context, throwOnFlattenableError(subobjects))
                         .forEach(m -> mergedMappers.put(m.leafName(), m));
                 } else {
                     putMergedMapper(mergedMappers, childOfExistingMapper);
@@ -638,11 +681,9 @@ public class ObjectMapper extends Mapper {
             for (Mapper mergeWithMapper : mergeWithObject) {
                 Mapper mergeIntoMapper = mergedMappers.get(mergeWithMapper.leafName());
                 if (mergeIntoMapper == null) {
-                    if (subobjects.isPresent()
-                        && subobjects.get() == Subobjects.DISABLED
-                        && mergeWithMapper instanceof ObjectMapper objectMapper) {
-                        // An existing mapping that has set `subobjects: false` is merged with a mapping with sub-objects
-                        objectMapper.asFlattenedFieldMappers(objectMergeContext.getMapperBuilderContext())
+                    if (mergeWithMapper instanceof ObjectMapper objectMapper && isFlatteningCandidate(subobjects, objectMapper)) {
+                        // An existing mapping with `subobjects` set to false or auto is merged with a mapping with sub-objects
+                        objectMapper.asFlattenedFieldMappers(context, throwOnFlattenableError(subobjects))
                             .stream()
                             .filter(m -> objectMergeContext.decrementFieldBudgetIfPossible(m.getTotalFieldsCount()))
                             .forEach(m -> putMergedMapper(mergedMappers, m));
@@ -699,57 +740,83 @@ public class ObjectMapper extends Mapper {
      *
      * @throws IllegalArgumentException if the mapper cannot be flattened
      */
-    List<FieldMapper> asFlattenedFieldMappers(MapperBuilderContext context) {
-        List<FieldMapper> flattenedMappers = new ArrayList<>();
+    List<Mapper> asFlattenedFieldMappers(MapperBuilderContext context, boolean throwOnFlattenableError) {
+        List<Mapper> flattenedMappers = new ArrayList<>();
         ContentPath path = new ContentPath();
-        asFlattenedFieldMappers(context, flattenedMappers, path);
+        asFlattenedFieldMappers(context, flattenedMappers, path, throwOnFlattenableError);
         return flattenedMappers;
     }
 
-    private void asFlattenedFieldMappers(MapperBuilderContext context, List<FieldMapper> flattenedMappers, ContentPath path) {
-        ensureFlattenable(context, path);
+    static boolean isFlatteningCandidate(Optional<Subobjects> subobjects, ObjectMapper mapper) {
+        return subobjects.isPresent() && subobjects.get() != Subobjects.ENABLED && mapper instanceof NestedObjectMapper == false;
+    }
+
+    private static boolean throwOnFlattenableError(Optional<Subobjects> subobjects) {
+        return subobjects.isPresent() && subobjects.get() == Subobjects.DISABLED;
+    }
+
+    private void asFlattenedFieldMappers(
+        MapperBuilderContext context,
+        List<Mapper> flattenedMappers,
+        ContentPath path,
+        boolean throwOnFlattenableError
+    ) {
+        var error = checkFlattenable(context);
+        if (error.isPresent()) {
+            if (throwOnFlattenableError) {
+                throw new IllegalArgumentException(
+                    "Object mapper ["
+                        + path.pathAsText(leafName())
+                        + "] was found in a context where subobjects is set to false. "
+                        + "Auto-flattening ["
+                        + path.pathAsText(leafName())
+                        + "] failed because "
+                        + error.get()
+                );
+            }
+            // The object can't be auto-flattened under the parent object, so it gets added at the current level.
+            // [subobjects=auto] applies auto-flattening to names, so the leaf name may need to change.
+            // Since mapper objects are immutable, we create a clone of the current one with the updated leaf name.
+            flattenedMappers.add(
+                path.pathAsText("").isEmpty()
+                    ? this
+                    : new ObjectMapper(path.pathAsText(leafName()), fullPath, enabled, subobjects, storeArraySource, dynamic, mappers)
+            );
+            return;
+        }
         path.add(leafName());
         for (Mapper mapper : mappers.values()) {
             if (mapper instanceof FieldMapper fieldMapper) {
                 FieldMapper.Builder fieldBuilder = fieldMapper.getMergeBuilder();
                 fieldBuilder.setLeafName(path.pathAsText(mapper.leafName()));
                 flattenedMappers.add(fieldBuilder.build(context));
-            } else if (mapper instanceof ObjectMapper objectMapper) {
-                objectMapper.asFlattenedFieldMappers(context, flattenedMappers, path);
+            } else if (mapper instanceof ObjectMapper objectMapper && mapper instanceof NestedObjectMapper == false) {
+                objectMapper.asFlattenedFieldMappers(context, flattenedMappers, path, throwOnFlattenableError);
             }
         }
         path.remove();
     }
 
-    private void ensureFlattenable(MapperBuilderContext context, ContentPath path) {
-        if (dynamic != null && context.getDynamic() != dynamic) {
-            throwAutoFlatteningException(
-                path,
+    Optional<String> checkFlattenable(MapperBuilderContext context) {
+        if (dynamic != null && (context == null || context.getDynamic() != dynamic)) {
+            return Optional.of(
                 "the value of [dynamic] ("
                     + dynamic
                     + ") is not compatible with the value from its parent context ("
-                    + context.getDynamic()
+                    + (context != null ? context.getDynamic() : "")
                     + ")"
             );
         }
+        if (storeArraySource()) {
+            return Optional.of("the value of [store_array_source] is [true]");
+        }
         if (isEnabled() == false) {
-            throwAutoFlatteningException(path, "the value of [enabled] is [false]");
+            return Optional.of("the value of [enabled] is [false]");
         }
-        if (subobjects.isPresent() && subobjects.get() == Subobjects.ENABLED) {
-            throwAutoFlatteningException(path, "the value of [subobjects] is [true]");
+        if (subobjects.isPresent() && subobjects.get() != Subobjects.DISABLED) {
+            return Optional.of("the value of [subobjects] is [" + subobjects().printedValue + "]");
         }
-    }
-
-    private void throwAutoFlatteningException(ContentPath path, String reason) {
-        throw new IllegalArgumentException(
-            "Object mapper ["
-                + path.pathAsText(leafName())
-                + "] was found in a context where subobjects is set to false. "
-                + "Auto-flattening ["
-                + path.pathAsText(leafName())
-                + "] failed because "
-                + reason
-        );
+        return Optional.empty();
     }
 
     @Override

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

@@ -2307,6 +2307,60 @@ public class DocumentParserTests extends MapperServiceTestCase {
         assertNotNull(doc.rootDoc().getField("attributes.simple.attribute"));
     }
 
+    public void testSubobjectsAutoFlattened() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(mapping(b -> {
+            b.startObject("attributes");
+            {
+                b.field("dynamic", false);
+                b.field("subobjects", "auto");
+                b.startObject("properties");
+                {
+                    b.startObject("simple.attribute").field("type", "keyword").endObject();
+                    b.startObject("complex.attribute").field("type", "flattened").endObject();
+                    b.startObject("path").field("type", "object");
+                    {
+                        b.field("store_array_source", "true").field("subobjects", "auto");
+                        b.startObject("properties");
+                        {
+                            b.startObject("nested.attribute").field("type", "keyword").endObject();
+                        }
+                        b.endObject();
+                    }
+                    b.endObject();
+                    b.startObject("flattened_object").field("type", "object");
+                    {
+                        b.startObject("properties");
+                        {
+                            b.startObject("nested.attribute").field("type", "keyword").endObject();
+                        }
+                        b.endObject();
+                    }
+                    b.endObject();
+                }
+                b.endObject();
+            }
+            b.endObject();
+        }));
+        ParsedDocument doc = mapper.parse(source("""
+            {
+              "attributes": {
+                "complex.attribute": {
+                  "foo" : "bar"
+                },
+                "simple.attribute": "sa",
+                "path": {
+                  "nested.attribute": "na"
+                },
+                "flattened_object.nested.attribute": "fna"
+              }
+            }
+            """));
+        assertNotNull(doc.rootDoc().getField("attributes.complex.attribute"));
+        assertNotNull(doc.rootDoc().getField("attributes.simple.attribute"));
+        assertNotNull(doc.rootDoc().getField("attributes.path.nested.attribute"));
+        assertNotNull(doc.rootDoc().getField("attributes.flattened_object.nested.attribute"));
+    }
+
     public void testWriteToFieldAlias() throws Exception {
         DocumentMapper mapper = createDocumentMapper(mapping(b -> {
             b.startObject("alias-field");

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

@@ -13,9 +13,12 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xcontent.json.JsonXContent;
+import org.hamcrest.Matchers;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -26,7 +29,7 @@ public class DotExpandingXContentParserTests extends ESTestCase {
         final ContentPath contentPath = new ContentPath();
         try (
             XContentParser inputParser = createParser(JsonXContent.jsonXContent, withDots);
-            XContentParser expandedParser = DotExpandingXContentParser.expandDots(inputParser, contentPath)
+            XContentParser expandedParser = DotExpandingXContentParser.expandDots(inputParser, contentPath, null)
         ) {
             expandedParser.allowDuplicateKeys(true);
 
@@ -37,7 +40,7 @@ public class DotExpandingXContentParserTests extends ESTestCase {
                 expectedParser.allowDuplicateKeys(true);
                 try (
                     var p = createParser(JsonXContent.jsonXContent, withDots);
-                    XContentParser actualParser = DotExpandingXContentParser.expandDots(p, contentPath)
+                    XContentParser actualParser = DotExpandingXContentParser.expandDots(p, contentPath, null)
                 ) {
                     XContentParser.Token currentToken;
                     while ((currentToken = actualParser.nextToken()) != null) {
@@ -127,7 +130,7 @@ public class DotExpandingXContentParserTests extends ESTestCase {
     public void testDotsCollapsingFlatPaths() throws IOException {
         ContentPath contentPath = new ContentPath();
         XContentParser parser = DotExpandingXContentParser.expandDots(createParser(JsonXContent.jsonXContent, """
-            {"metrics.service.time": 10, "metrics.service.time.max": 500, "metrics.foo": "value"}"""), contentPath);
+            {"metrics.service.time": 10, "metrics.service.time.max": 500, "metrics.foo": "value"}"""), contentPath, null);
         parser.nextToken();
         assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
         assertEquals("metrics", parser.currentName());
@@ -197,7 +200,7 @@ public class DotExpandingXContentParserTests extends ESTestCase {
                 },
                 "foo" : "value"
               }
-            }"""), contentPath);
+            }"""), contentPath, null);
         parser.nextToken();
         assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
         assertEquals("metrics", parser.currentName());
@@ -235,7 +238,7 @@ public class DotExpandingXContentParserTests extends ESTestCase {
 
     public void testSkipChildren() throws IOException {
         XContentParser parser = DotExpandingXContentParser.expandDots(createParser(JsonXContent.jsonXContent, """
-            { "test.with.dots" : "value", "nodots" : "value2" }"""), new ContentPath());
+            { "test.with.dots" : "value", "nodots" : "value2" }"""), new ContentPath(), null);
         parser.nextToken();     // start object
         assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
         assertEquals("test", parser.currentName());
@@ -258,7 +261,7 @@ public class DotExpandingXContentParserTests extends ESTestCase {
 
     public void testSkipChildrenWithinInnerObject() throws IOException {
         XContentParser parser = DotExpandingXContentParser.expandDots(createParser(JsonXContent.jsonXContent, """
-            { "test.with.dots" : {"obj" : {"field":"value"}}, "nodots" : "value2" }"""), new ContentPath());
+            { "test.with.dots" : {"obj" : {"field":"value"}}, "nodots" : "value2" }"""), new ContentPath(), null);
 
         parser.nextToken();     // start object
         assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
@@ -306,7 +309,8 @@ public class DotExpandingXContentParserTests extends ESTestCase {
         XContentParser expectedParser = createParser(JsonXContent.jsonXContent, jsonInput);
         XContentParser dotExpandedParser = DotExpandingXContentParser.expandDots(
             createParser(JsonXContent.jsonXContent, jsonInput),
-            new ContentPath()
+            new ContentPath(),
+            null
         );
 
         assertEquals(expectedParser.getTokenLocation(), dotExpandedParser.getTokenLocation());
@@ -364,7 +368,8 @@ public class DotExpandingXContentParserTests extends ESTestCase {
     public void testParseMapUOE() throws Exception {
         XContentParser dotExpandedParser = DotExpandingXContentParser.expandDots(
             createParser(JsonXContent.jsonXContent, ""),
-            new ContentPath()
+            new ContentPath(),
+            null
         );
         expectThrows(UnsupportedOperationException.class, dotExpandedParser::map);
     }
@@ -372,7 +377,8 @@ public class DotExpandingXContentParserTests extends ESTestCase {
     public void testParseMapOrderedUOE() throws Exception {
         XContentParser dotExpandedParser = DotExpandingXContentParser.expandDots(
             createParser(JsonXContent.jsonXContent, ""),
-            new ContentPath()
+            new ContentPath(),
+            null
         );
         expectThrows(UnsupportedOperationException.class, dotExpandedParser::mapOrdered);
     }
@@ -380,7 +386,8 @@ public class DotExpandingXContentParserTests extends ESTestCase {
     public void testParseMapStringsUOE() throws Exception {
         XContentParser dotExpandedParser = DotExpandingXContentParser.expandDots(
             createParser(JsonXContent.jsonXContent, ""),
-            new ContentPath()
+            new ContentPath(),
+            null
         );
         expectThrows(UnsupportedOperationException.class, dotExpandedParser::mapStrings);
     }
@@ -388,7 +395,8 @@ public class DotExpandingXContentParserTests extends ESTestCase {
     public void testParseMapSupplierUOE() throws Exception {
         XContentParser dotExpandedParser = DotExpandingXContentParser.expandDots(
             createParser(JsonXContent.jsonXContent, ""),
-            new ContentPath()
+            new ContentPath(),
+            null
         );
         expectThrows(UnsupportedOperationException.class, () -> dotExpandedParser.map(HashMap::new, XContentParser::text));
     }
@@ -403,7 +411,8 @@ public class DotExpandingXContentParserTests extends ESTestCase {
         contentPath.setWithinLeafObject(true);
         XContentParser dotExpandedParser = DotExpandingXContentParser.expandDots(
             createParser(JsonXContent.jsonXContent, jsonInput),
-            contentPath
+            contentPath,
+            null
         );
         assertEquals(XContentParser.Token.START_OBJECT, dotExpandedParser.nextToken());
         assertEquals(XContentParser.Token.FIELD_NAME, dotExpandedParser.nextToken());
@@ -418,7 +427,8 @@ public class DotExpandingXContentParserTests extends ESTestCase {
     public void testParseListUOE() throws Exception {
         XContentParser dotExpandedParser = DotExpandingXContentParser.expandDots(
             createParser(JsonXContent.jsonXContent, ""),
-            new ContentPath()
+            new ContentPath(),
+            null
         );
         expectThrows(UnsupportedOperationException.class, dotExpandedParser::list);
     }
@@ -426,7 +436,8 @@ public class DotExpandingXContentParserTests extends ESTestCase {
     public void testParseListOrderedUOE() throws Exception {
         XContentParser dotExpandedParser = DotExpandingXContentParser.expandDots(
             createParser(JsonXContent.jsonXContent, ""),
-            new ContentPath()
+            new ContentPath(),
+            null
         );
         expectThrows(UnsupportedOperationException.class, dotExpandedParser::listOrderedMap);
     }
@@ -440,7 +451,8 @@ public class DotExpandingXContentParserTests extends ESTestCase {
         contentPath.setWithinLeafObject(true);
         XContentParser dotExpandedParser = DotExpandingXContentParser.expandDots(
             createParser(JsonXContent.jsonXContent, jsonInput),
-            contentPath
+            contentPath,
+            null
         );
         assertEquals(XContentParser.Token.START_OBJECT, dotExpandedParser.nextToken());
         assertEquals(XContentParser.Token.FIELD_NAME, dotExpandedParser.nextToken());
@@ -450,4 +462,104 @@ public class DotExpandingXContentParserTests extends ESTestCase {
         assertEquals("one", list.get(0));
         assertEquals("two", list.get(1));
     }
+
+    private static DocumentParserContext createContext(XContentBuilder builder) throws IOException {
+        var documentMapper = new MapperServiceTestCase() {
+        }.createDocumentMapper(builder);
+        return new TestDocumentParserContext(documentMapper.mappers(), null);
+    }
+
+    private static List<String> getSubPaths(XContentBuilder builder, String... path) throws IOException {
+        DocumentParserContext context = createContext(builder);
+        return DotExpandingXContentParser.maybeFlattenPaths(Arrays.stream(path).toList(), context, new ContentPath());
+    }
+
+    private static List<String> getSubPaths(XContentBuilder builder, List<String> contentPath, List<String> path) throws IOException {
+        DocumentParserContext context = createContext(builder);
+        ContentPath content = new ContentPath();
+        for (String c : contentPath) {
+            content.add(c);
+        }
+        return DotExpandingXContentParser.maybeFlattenPaths(path, context, content);
+    }
+
+    public void testAutoFlattening() throws Exception {
+        var b = XContentBuilder.builder(XContentType.JSON.xContent());
+        b.startObject().startObject("_doc");
+        {
+            b.field("subobjects", "auto");
+            b.startObject("properties");
+            {
+                b.startObject("path").startObject("properties");
+                {
+                    b.startObject("to").startObject("properties");
+                    {
+                        b.startObject("field").field("type", "integer").endObject();
+                    }
+                    b.endObject().endObject();
+                }
+                b.endObject().endObject();
+                b.startObject("path.auto").field("subobjects", "auto").startObject("properties");
+                {
+                    b.startObject("to").startObject("properties");
+                    {
+                        b.startObject("some.field").field("type", "integer").endObject();
+                    }
+                    b.endObject().endObject();
+                    b.startObject("inner.enabled").field("dynamic", "false").startObject("properties");
+                    {
+                        b.startObject("field").field("type", "integer").endObject();
+                    }
+                    b.endObject().endObject();
+                }
+                b.endObject().endObject();
+                b.startObject("path.disabled").field("subobjects", "false").startObject("properties");
+                {
+                    b.startObject("to").startObject("properties");
+                    {
+                        b.startObject("some.field").field("type", "integer").endObject();
+                    }
+                    b.endObject().endObject();
+                }
+                b.endObject().endObject();
+            }
+            b.endObject();
+        }
+        b.endObject().endObject();
+
+        // inner [subobjects:enabled] gets flattened
+        assertThat(getSubPaths(b, "field"), Matchers.contains("field"));
+        assertThat(getSubPaths(b, "path", "field"), Matchers.contains("path.field"));
+        assertThat(getSubPaths(b, "path", "to", "field"), Matchers.contains("path.to.field"));
+        assertThat(getSubPaths(b, "path", "to", "any"), Matchers.contains("path.to.any"));
+
+        // inner [subobjects:auto] does not get flattened
+        assertThat(getSubPaths(b, "path", "auto", "field"), Matchers.contains("path.auto", "field"));
+        assertThat(getSubPaths(b, "path", "auto", "some", "field"), Matchers.contains("path.auto", "some.field"));
+        assertThat(getSubPaths(b, "path", "auto", "to", "some", "field"), Matchers.contains("path.auto", "to.some.field"));
+        assertThat(getSubPaths(b, "path", "auto", "to", "some", "other"), Matchers.contains("path.auto", "to.some.other"));
+        assertThat(getSubPaths(b, "path", "auto", "inner", "enabled", "field"), Matchers.contains("path.auto", "inner.enabled", "field"));
+        assertThat(
+            getSubPaths(b, "path", "auto", "inner", "enabled", "to", "some", "field"),
+            Matchers.contains("path.auto", "inner.enabled", "to", "some", "field")
+        );
+
+        // inner [subobjects:disabled] gets flattened
+        assertThat(getSubPaths(b, "path", "disabled", "field"), Matchers.contains("path.disabled.field"));
+        assertThat(getSubPaths(b, "path", "disabled", "some", "field"), Matchers.contains("path.disabled.some.field"));
+        assertThat(getSubPaths(b, "path", "disabled", "to", "some", "field"), Matchers.contains("path.disabled.to.some.field"));
+        assertThat(getSubPaths(b, "path", "disabled", "to", "some", "other"), Matchers.contains("path.disabled.to.some.other"));
+
+        // Non-empty content path.
+        assertThat(getSubPaths(b, List.of("path"), List.of("field")), Matchers.contains("field"));
+        assertThat(getSubPaths(b, List.of("path"), List.of("to", "field")), Matchers.contains("to", "field"));
+        assertThat(getSubPaths(b, List.of("path", "to"), List.of("field")), Matchers.contains("field"));
+        assertThat(getSubPaths(b, List.of("path"), List.of("auto", "field")), Matchers.contains("auto", "field"));
+        assertThat(getSubPaths(b, List.of("path", "auto"), List.of("to", "some", "field")), Matchers.contains("to.some.field"));
+        assertThat(
+            getSubPaths(b, List.of("path", "auto"), List.of("inner", "enabled", "to", "some", "field")),
+            Matchers.contains("inner.enabled", "to", "some", "field")
+        );
+        assertThat(getSubPaths(b, List.of("path", "disabled"), List.of("to", "some", "field")), Matchers.contains("to", "some", "field"));
+    }
 }

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

@@ -1619,10 +1619,9 @@ public class DynamicTemplatesTests extends MapperServiceTestCase {
 
         assertNotNull(doc.rootDoc().get("metrics.time.max"));
         assertNotNull(doc.docs().get(0).get("metrics.time.foo"));
-        assertThat(
-            ((ObjectMapper) doc.dynamicMappingsUpdate().getRoot().getMapper("metrics")).getMapper("time"),
-            instanceOf(NestedObjectMapper.class)
-        );
+        var metrics = ((ObjectMapper) doc.dynamicMappingsUpdate().getRoot().getMapper("metrics"));
+        assertThat(metrics.getMapper("time"), instanceOf(NestedObjectMapper.class));
+        assertThat(metrics.getMapper("time.max"), instanceOf(NumberFieldMapper.class));
     }
 
     public void testDynamicSubobject() throws IOException {
@@ -2057,7 +2056,7 @@ public class DynamicTemplatesTests extends MapperServiceTestCase {
                 "dynamic_templates": [
                   {
                     "test": {
-                      "path_match": "attributes.resource.*",
+                      "path_match": "attributes.*",
                       "match_mapping_type": "object",
                       "mapping": {
                         "type": "flattened"
@@ -2070,7 +2069,7 @@ public class DynamicTemplatesTests extends MapperServiceTestCase {
             """;
         String docJson = """
             {
-              "attributes.resource": {
+              "attributes": {
                 "complex.attribute": {
                   "a": "b"
                 },
@@ -2083,14 +2082,67 @@ public class DynamicTemplatesTests extends MapperServiceTestCase {
         ParsedDocument parsedDoc = mapperService.documentMapper().parse(source(docJson));
         merge(mapperService, dynamicMapping(parsedDoc.dynamicMappingsUpdate()));
 
-        Mapper fooBarMapper = mapperService.documentMapper().mappers().getMapper("attributes.resource.foo.bar");
+        Mapper fooBarMapper = mapperService.documentMapper().mappers().getMapper("attributes.foo.bar");
         assertNotNull(fooBarMapper);
         assertEquals("text", fooBarMapper.typeName());
-        Mapper fooStructuredMapper = mapperService.documentMapper().mappers().getMapper("attributes.resource.complex.attribute");
+        Mapper fooStructuredMapper = mapperService.documentMapper().mappers().getMapper("attributes.complex.attribute");
         assertNotNull(fooStructuredMapper);
         assertEquals("flattened", fooStructuredMapper.typeName());
     }
 
+    public void testSubobjectsAutoWithObjectInDynamicTemplate() throws IOException {
+        String mapping = """
+            {
+              "_doc": {
+                "properties": {
+                  "attributes": {
+                    "type": "object",
+                    "subobjects": "auto"
+                  }
+                },
+                "dynamic_templates": [
+                  {
+                    "test": {
+                      "path_match": "attributes.*",
+                      "match_mapping_type": "object",
+                      "mapping": {
+                        "type": "object",
+                        "dynamic": "false",
+                        "properties": {
+                          "id": {
+                            "type": "integer"
+                          }
+                        }
+                      }
+                    }
+                  }
+                ]
+              }
+            }
+            """;
+        String docJson = """
+            {
+              "attributes": {
+                "to": {
+                  "id": 10
+                },
+                "foo.bar": "baz"
+              }
+            }
+            """;
+
+        MapperService mapperService = createMapperService(mapping);
+        ParsedDocument parsedDoc = mapperService.documentMapper().parse(source(docJson));
+        merge(mapperService, dynamicMapping(parsedDoc.dynamicMappingsUpdate()));
+
+        Mapper fooBarMapper = mapperService.documentMapper().mappers().getMapper("attributes.foo.bar");
+        assertNotNull(fooBarMapper);
+        assertEquals("text", fooBarMapper.typeName());
+        Mapper innerObject = mapperService.documentMapper().mappers().objectMappers().get("attributes.to");
+        assertNotNull(innerObject);
+        assertEquals("integer", mapperService.documentMapper().mappers().getMapper("attributes.to.id").typeName());
+    }
+
     public void testMatchWithArrayOfFieldNames() throws IOException {
         String mapping = """
             {

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

@@ -1549,6 +1549,66 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
         assertEquals("{\"path\":{\"at\":\"A\"}}", syntheticSource);
     }
 
+    public void testCopyToRootWithSubobjectFlattening() throws IOException {
+        DocumentMapper documentMapper = createMapperService(topMapping(b -> {
+            b.startObject("_source").field("mode", "synthetic").endObject();
+            b.field("subobjects", randomFrom("false", "auto"));
+            b.startObject("properties");
+            {
+                b.startObject("k").field("type", "keyword").field("copy_to", "a.b.c").endObject();
+                b.startObject("a").startObject("properties");
+                {
+                    b.startObject("b").startObject("properties");
+                    {
+                        b.startObject("c").field("type", "keyword").endObject();
+                    }
+                    b.endObject().endObject();
+                }
+                b.endObject().endObject();
+            }
+            b.endObject();
+        })).documentMapper();
+
+        CheckedConsumer<XContentBuilder, IOException> document = b -> b.field("k", "hey");
+
+        var doc = documentMapper.parse(source(document));
+        assertNotNull(doc.docs().get(0).getField("a.b.c"));
+
+        var syntheticSource = syntheticSource(documentMapper, document);
+        assertEquals("{\"k\":\"hey\"}", syntheticSource);
+    }
+
+    public void testCopyToObjectWithSubobjectFlattening() throws IOException {
+        DocumentMapper documentMapper = createMapperService(syntheticSourceMapping(b -> {
+            b.startObject("path").field("subobjects", randomFrom("false", "auto")).startObject("properties");
+            {
+                b.startObject("k").field("type", "keyword").field("copy_to", "path.a.b.c").endObject();
+                b.startObject("a").startObject("properties");
+                {
+                    b.startObject("b").startObject("properties");
+                    {
+                        b.startObject("c").field("type", "keyword").endObject();
+                    }
+                    b.endObject().endObject();
+                }
+                b.endObject().endObject();
+            }
+            b.endObject().endObject();
+        })).documentMapper();
+
+        CheckedConsumer<XContentBuilder, IOException> document = b -> {
+            b.startObject("path");
+            b.field("k", "hey");
+            b.endObject();
+        };
+
+        var doc = documentMapper.parse(source(document));
+        assertNotNull(doc.docs().get(0).getField("path.a.b.c"));
+
+        var syntheticSource = syntheticSource(documentMapper, document);
+        assertEquals("{\"path\":{\"k\":\"hey\"}}", syntheticSource);
+    }
+
     protected void validateRoundTripReader(String syntheticSource, DirectoryReader reader, DirectoryReader roundTripReader)
         throws IOException {
         // We exclude ignored source field since in some cases it contains an exact copy of a part of document source.

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

@@ -354,12 +354,8 @@ public class ObjectMapperTests extends MapperServiceTestCase {
                 b.field("subobjects", false);
                 b.startObject("properties");
                 {
-                    b.startObject("time");
-                    b.field("type", "long");
-                    b.endObject();
-                    b.startObject("time.max");
-                    b.field("type", "long");
-                    b.endObject();
+                    b.startObject("time").field("type", "long").endObject();
+                    b.startObject("time.max").field("type", "long").endObject();
                 }
                 b.endObject();
             }
@@ -380,9 +376,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
                     {
                         b.startObject("properties");
                         {
-                            b.startObject("max");
-                            b.field("type", "long");
-                            b.endObject();
+                            b.startObject("max").field("type", "long").endObject();
                         }
                         b.endObject();
                     }
@@ -403,9 +397,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
                 b.field("subobjects", false);
                 b.startObject("properties");
                 {
-                    b.startObject("time");
-                    b.field("type", "nested");
-                    b.endObject();
+                    b.startObject("time").field("type", "nested").endObject();
                 }
                 b.endObject();
             }
@@ -419,12 +411,8 @@ public class ObjectMapperTests extends MapperServiceTestCase {
 
     public void testSubobjectsFalseRoot() throws Exception {
         MapperService mapperService = createMapperService(mappingNoSubobjects(b -> {
-            b.startObject("metrics.service.time");
-            b.field("type", "long");
-            b.endObject();
-            b.startObject("metrics.service.time.max");
-            b.field("type", "long");
-            b.endObject();
+            b.startObject("metrics.service.time").field("type", "long").endObject();
+            b.startObject("metrics.service.time.max").field("type", "long").endObject();
         }));
         assertNotNull(mapperService.fieldType("metrics.service.time"));
         assertNotNull(mapperService.fieldType("metrics.service.time.max"));
@@ -441,9 +429,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
             {
                 b.startObject("properties");
                 {
-                    b.startObject("max");
-                    b.field("type", "long");
-                    b.endObject();
+                    b.startObject("max").field("type", "long").endObject();
                 }
                 b.endObject();
             }
@@ -455,9 +441,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
 
     public void testSubobjectsFalseRootWithInnerNested() {
         MapperParsingException exception = expectThrows(MapperParsingException.class, () -> createMapperService(mappingNoSubobjects(b -> {
-            b.startObject("metrics.service");
-            b.field("type", "nested");
-            b.endObject();
+            b.startObject("metrics.service").field("type", "nested").endObject();
         })));
         assertEquals(
             "Failed to parse mapping: Tried to add nested object [metrics.service] to object [_doc] which does not support subobjects",
@@ -473,8 +457,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
             "_doc",
             MergeReason.MAPPING_UPDATE,
             new CompressedXContent(BytesReference.bytes(fieldMapping(b -> {
-                b.field("type", "object");
-                b.field("subobjects", "false");
+                b.field("type", "object").field("subobjects", "false");
             })))
         );
         MapperException exception = expectThrows(
@@ -509,12 +492,8 @@ public class ObjectMapperTests extends MapperServiceTestCase {
                 b.field("subobjects", "auto");
                 b.startObject("properties");
                 {
-                    b.startObject("time");
-                    b.field("type", "long");
-                    b.endObject();
-                    b.startObject("time.max");
-                    b.field("type", "long");
-                    b.endObject();
+                    b.startObject("time").field("type", "long").endObject();
+                    b.startObject("time.max").field("type", "long").endObject();
                     b.startObject("attributes");
                     {
                         b.field("type", "object");
@@ -531,7 +510,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
         assertNotNull(mapperService.documentMapper().mappers().objectMappers().get("metrics.service.attributes"));
     }
 
-    public void testSubobjectsAutoWithInnerObject() throws IOException {
+    public void testSubobjectsAutoWithInnerFlattenableObject() throws IOException {
         MapperService mapperService = createMapperService(mapping(b -> {
             b.startObject("metrics.service");
             {
@@ -542,16 +521,12 @@ public class ObjectMapperTests extends MapperServiceTestCase {
                     {
                         b.startObject("properties");
                         {
-                            b.startObject("max");
-                            b.field("type", "long");
-                            b.endObject();
+                            b.startObject("max").field("type", "long").endObject();
                         }
                         b.endObject();
                     }
                     b.endObject();
-                    b.startObject("foo");
-                    b.field("type", "keyword");
-                    b.endObject();
+                    b.startObject("foo").field("type", "keyword").endObject();
                 }
                 b.endObject();
             }
@@ -560,7 +535,37 @@ public class ObjectMapperTests extends MapperServiceTestCase {
         assertNull(mapperService.fieldType("metrics.service.time"));
         assertNotNull(mapperService.fieldType("metrics.service.time.max"));
         assertNotNull(mapperService.fieldType("metrics.service.foo"));
-        assertNotNull(mapperService.documentMapper().mappers().objectMappers().get("metrics.service.time"));
+        assertNull(mapperService.documentMapper().mappers().objectMappers().get("metrics.service.time"));  // Gets flattened.
+        assertNotNull(mapperService.documentMapper().mappers().getMapper("metrics.service.foo"));
+    }
+
+    public void testSubobjectsAutoWithInnerNonFlattenableObject() throws IOException {
+        MapperService mapperService = createMapperService(mapping(b -> {
+            b.startObject("metrics.service");
+            {
+                b.field("subobjects", "auto");
+                b.startObject("properties");
+                {
+                    b.startObject("time");
+                    {
+                        b.field(ObjectMapper.STORE_ARRAY_SOURCE_PARAM, true);
+                        b.startObject("properties");
+                        {
+                            b.startObject("max").field("type", "long").endObject();
+                        }
+                        b.endObject();
+                    }
+                    b.endObject();
+                    b.startObject("foo").field("type", "keyword").endObject();
+                }
+                b.endObject();
+            }
+            b.endObject();
+        }));
+        assertNull(mapperService.fieldType("metrics.service.time"));
+        assertNotNull(mapperService.fieldType("metrics.service.time.max"));
+        assertNotNull(mapperService.fieldType("metrics.service.foo"));
+        assertNotNull(mapperService.documentMapper().mappers().objectMappers().get("metrics.service.time"));  // Not flattened.
         assertNotNull(mapperService.documentMapper().mappers().getMapper("metrics.service.foo"));
     }
 
@@ -571,9 +576,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
                 b.field("subobjects", "auto");
                 b.startObject("properties");
                 {
-                    b.startObject("time");
-                    b.field("type", "nested");
-                    b.endObject();
+                    b.startObject("time").field("type", "nested").endObject();
                 }
                 b.endObject();
             }
@@ -587,12 +590,8 @@ public class ObjectMapperTests extends MapperServiceTestCase {
 
     public void testSubobjectsAutoRoot() throws Exception {
         MapperService mapperService = createMapperService(mappingWithSubobjects(b -> {
-            b.startObject("metrics.service.time");
-            b.field("type", "long");
-            b.endObject();
-            b.startObject("metrics.service.time.max");
-            b.field("type", "long");
-            b.endObject();
+            b.startObject("metrics.service.time").field("type", "long").endObject();
+            b.startObject("metrics.service.time.max").field("type", "long").endObject();
             b.startObject("metrics.attributes");
             {
                 b.field("type", "object");
@@ -605,15 +604,13 @@ public class ObjectMapperTests extends MapperServiceTestCase {
         assertNotNull(mapperService.documentMapper().mappers().objectMappers().get("metrics.attributes"));
     }
 
-    public void testSubobjectsAutoRootWithInnerObject() throws IOException {
+    public void testSubobjectsAutoRootWithInnerFlattenableObject() throws IOException {
         MapperService mapperService = createMapperService(mappingWithSubobjects(b -> {
             b.startObject("metrics.service.time");
             {
                 b.startObject("properties");
                 {
-                    b.startObject("max");
-                    b.field("type", "long");
-                    b.endObject();
+                    b.startObject("max").field("type", "long").endObject();
                 }
                 b.endObject();
             }
@@ -621,8 +618,48 @@ public class ObjectMapperTests extends MapperServiceTestCase {
         }, "auto"));
         assertNull(mapperService.fieldType("metrics.service.time"));
         assertNotNull(mapperService.fieldType("metrics.service.time.max"));
-        assertNotNull(mapperService.documentMapper().mappers().objectMappers().get("metrics.service.time"));
-        assertNotNull(mapperService.documentMapper().mappers().getMapper("metrics.service.time.max"));
+        assertNull(mapperService.documentMapper().mappers().objectMappers().get("metrics.service.time"));  // Gets flattened.
+
+        Mapper innerField = mapperService.documentMapper().mappers().getMapper("metrics.service.time.max");
+        assertNotNull(innerField);
+        assertEquals("metrics.service.time.max", innerField.leafName());
+    }
+
+    public void testSubobjectsAutoRootWithInnerNonFlattenableObject() throws IOException {
+        MapperService mapperService = createMapperService(mappingWithSubobjects(b -> {
+            b.startObject("metrics").startObject("properties");
+            {
+                b.startObject("service.time");
+                {
+                    b.field("subobjects", "auto");
+                    b.startObject("properties");
+                    {
+                        b.startObject("path").startObject("properties");
+                        {
+                            b.startObject("to").startObject("properties");
+                            {
+                                b.startObject("max").field("type", "long").endObject();
+                            }
+                            b.endObject().endObject();
+                        }
+                        b.endObject().endObject();
+                    }
+                    b.endObject();
+                }
+                b.endObject();
+            }
+            b.endObject().endObject();
+        }, "auto"));
+        assertNull(mapperService.fieldType("metrics.service.time"));
+        assertNotNull(mapperService.fieldType("metrics.service.time.path.to.max"));
+
+        ObjectMapper innerObject = mapperService.documentMapper().mappers().objectMappers().get("metrics.service.time");  // Not flattened.
+        assertNotNull(innerObject);
+        assertEquals("metrics.service.time", innerObject.leafName());
+
+        Mapper innerField = mapperService.documentMapper().mappers().getMapper("metrics.service.time.path.to.max");
+        assertNotNull(innerField);
+        assertEquals("path.to.max", innerField.leafName());
     }
 
     public void testSubobjectsAutoRootWithInnerNested() throws IOException {
@@ -742,16 +779,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
         ObjectMapper objectMapper = new ObjectMapper.Builder("parent", Optional.empty()).add(
             new ObjectMapper.Builder("child", Optional.empty()).add(new KeywordFieldMapper.Builder("keyword2", IndexVersion.current()))
         ).add(new KeywordFieldMapper.Builder("keyword1", IndexVersion.current())).build(rootContext);
-        List<String> fields = objectMapper.asFlattenedFieldMappers(rootContext).stream().map(FieldMapper::fullPath).toList();
-        assertThat(fields, containsInAnyOrder("parent.keyword1", "parent.child.keyword2"));
-    }
-
-    public void testFlattenSubobjectsAuto() {
-        MapperBuilderContext rootContext = MapperBuilderContext.root(false, false);
-        ObjectMapper objectMapper = new ObjectMapper.Builder("parent", Optional.of(ObjectMapper.Subobjects.AUTO)).add(
-            new ObjectMapper.Builder("child", Optional.empty()).add(new KeywordFieldMapper.Builder("keyword2", IndexVersion.current()))
-        ).add(new KeywordFieldMapper.Builder("keyword1", IndexVersion.current())).build(rootContext);
-        List<String> fields = objectMapper.asFlattenedFieldMappers(rootContext).stream().map(FieldMapper::fullPath).toList();
+        List<String> fields = objectMapper.asFlattenedFieldMappers(rootContext, true).stream().map(Mapper::fullPath).toList();
         assertThat(fields, containsInAnyOrder("parent.keyword1", "parent.child.keyword2"));
     }
 
@@ -760,7 +788,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
         ObjectMapper objectMapper = new ObjectMapper.Builder("parent", Optional.of(ObjectMapper.Subobjects.DISABLED)).add(
             new ObjectMapper.Builder("child", Optional.empty()).add(new KeywordFieldMapper.Builder("keyword2", IndexVersion.current()))
         ).add(new KeywordFieldMapper.Builder("keyword1", IndexVersion.current())).build(rootContext);
-        List<String> fields = objectMapper.asFlattenedFieldMappers(rootContext).stream().map(FieldMapper::fullPath).toList();
+        List<String> fields = objectMapper.asFlattenedFieldMappers(rootContext, true).stream().map(Mapper::fullPath).toList();
         assertThat(fields, containsInAnyOrder("parent.keyword1", "parent.child.keyword2"));
     }
 
@@ -772,7 +800,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
 
         IllegalArgumentException exception = expectThrows(
             IllegalArgumentException.class,
-            () -> objectMapper.asFlattenedFieldMappers(rootContext)
+            () -> objectMapper.asFlattenedFieldMappers(rootContext, true)
         );
         assertEquals(
             "Object mapper [parent.child] was found in a context where subobjects is set to false. "
@@ -788,7 +816,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
 
         IllegalArgumentException exception = expectThrows(
             IllegalArgumentException.class,
-            () -> objectMapper.asFlattenedFieldMappers(rootContext)
+            () -> objectMapper.asFlattenedFieldMappers(rootContext, true)
         );
         assertEquals(
             "Object mapper [parent] was found in a context where subobjects is set to false. "
@@ -797,13 +825,30 @@ public class ObjectMapperTests extends MapperServiceTestCase {
         );
     }
 
+    public void testFlattenSubobjectsAuto() {
+        MapperBuilderContext rootContext = MapperBuilderContext.root(false, false);
+        ObjectMapper objectMapper = new ObjectMapper.Builder("parent", Optional.of(ObjectMapper.Subobjects.AUTO)).add(
+            new ObjectMapper.Builder("child", Optional.empty()).add(new KeywordFieldMapper.Builder("keyword2", IndexVersion.current()))
+        ).add(new KeywordFieldMapper.Builder("keyword1", IndexVersion.current())).build(rootContext);
+
+        IllegalArgumentException exception = expectThrows(
+            IllegalArgumentException.class,
+            () -> objectMapper.asFlattenedFieldMappers(rootContext, true)
+        );
+        assertEquals(
+            "Object mapper [parent] was found in a context where subobjects is set to false. "
+                + "Auto-flattening [parent] failed because the value of [subobjects] is [auto]",
+            exception.getMessage()
+        );
+    }
+
     public void testFlattenExplicitSubobjectsTrue() {
         MapperBuilderContext rootContext = MapperBuilderContext.root(false, false);
         ObjectMapper objectMapper = new ObjectMapper.Builder("parent", Optional.of(ObjectMapper.Subobjects.ENABLED)).build(rootContext);
 
         IllegalArgumentException exception = expectThrows(
             IllegalArgumentException.class,
-            () -> objectMapper.asFlattenedFieldMappers(rootContext)
+            () -> objectMapper.asFlattenedFieldMappers(rootContext, true)
         );
         assertEquals(
             "Object mapper [parent] was found in a context where subobjects is set to false. "