Add support for dots in field names for metrics usecases (#86166)

This PR adds support for a new mapping parameter to the configuration of the object mapper (root as well as individual fields), that makes it possible to store metrics data where it's common to have fields with dots in their names in the following format:

```
{
  "metrics.time" : 10,
  "metrics.time.min" : 1,
  "metrics.time.max" : 500
}
```

Instead of expanding dotted paths the their corresponding object structure, objects can be configured to preserve dots in field names, in which case they can only hold leaf sub-fields and no further objects.

The mapping parameter is called subobjects and controls whether an object can hold other objects (defaults to true) or not. The following example shows how it can be configured in the mappings:

```
{
  "mappings" : {
    "properties" : {
      "metrics" : {
        "type" : "object", 
        "subobjects" : false
      }
    }
  }
}
```

Closes #63530

docs/changelog/86166.yaml

@@ -0,0 +1,48 @@
+pr: 86166
+summary: Add support for dots in field names for metrics usecases
+area: Mapping
+type: feature
+issues:
+ - 63530
+highlight:
+  title: Add support for dots in field names for metrics usecases
+  body: |-
+    Metrics data can often be made of several fields with dots in their names,
+    sharing common prefixes, like in the following example:
+
+    ```
+    {
+      "metrics.time" : 10,
+      "metrics.time.min" : 1,
+      "metrics.time.max" : 500
+    }
+    ```
+
+    Such format causes a mapping conflict as the `metrics.time` holds a value,
+    but it also needs to be mapped as an object in order to hold the `min` and
+    `max` leaf fields.
+
+    A new object mapping parameter called `subobjects`, which defaults to `true`,
+    has been introduced to preserve dots in field names. An object with `subobjects`
+    set to `false` can only ever hold leaf sub-fields and no further objects. The
+    following example shows how it can be configured in the mappings for the
+    `metrics` object:
+
+    ```
+    {
+      "mappings": {
+        "properties" : {
+          "metrics" : {
+            "type" : "object",
+            "subobjects" : false
+          }
+        }
+      }
+    }
+    ```
+
+    With this configuration any child of `metrics` will be mapped unchanged,
+    without expanding dots in field names to the corresponding object structure.
+    That makes it possible to store the metrics document above.
+
+  notable: true

docs/reference/mapping/params.asciidoc

@@ -31,6 +31,7 @@ The following mapping parameters are common to some or all field data types:
 * <<properties,`properties`>>
 * <<search-analyzer,`search_analyzer`>>
 * <<similarity,`similarity`>>
+* <<subobjects,`subobjects`>>
 * <<mapping-store,`store`>>
 * <<term-vector,`term_vector`>>
 
@@ -83,4 +84,6 @@ include::params/similarity.asciidoc[]
 
 include::params/store.asciidoc[]
 
+include::params/subobjects.asciidoc[]
+
 include::params/term-vector.asciidoc[]

docs/reference/mapping/params/subobjects.asciidoc

@@ -0,0 +1,107 @@
+[[subobjects]]
+=== `subobjects`
+
+When indexing a document or updating mappings, Elasticsearch accepts fields that contain dots in their names,
+which get expanded to their corresponding object structure. For instance, the  field `metrics.time.max`
+is mapped as a `max` leaf field with a parent `time` object, belonging to its parent `metrics` object.
+
+The described default behaviour is reasonable for most scenarios, but causes problems in certain situations
+where for instance a field `metrics.time` holds a value too, which is common when indexing metrics data.
+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
+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
+without the need for any intermediate object, as dots in field names are preserved.
+
+[source,console]
+--------------------------------------------------
+PUT my-index-000001
+{
+  "mappings": {
+    "properties": {
+      "metrics": {
+        "type":  "object",
+        "subobjects": false <1>
+      }
+    }
+  }
+}
+
+PUT my-index-000001/_doc/metric_1
+{
+  "metrics.time" : 100, <2>
+  "metrics.time.min" : 10,
+  "metrics.time.max" : 900
+}
+
+PUT my-index-000001/_doc/metric_2
+{
+  "metrics" : {
+    "time" : 100, <3>
+    "time.min" : 10,
+    "time.max" : 900
+  }
+}
+
+GET my-index-000001/_mapping
+--------------------------------------------------
+
+[source,console-result]
+--------------------------------------------------
+{
+  "my-index-000001" : {
+    "mappings" : {
+      "properties" : {
+        "metrics" : {
+          "subobjects" : false,
+          "properties" : {
+            "time" : {
+              "type" : "long"
+            },
+            "time.min" : { <4>
+              "type" : "long"
+            },
+            "time.max" : {
+              "type" : "long"
+            }
+          }
+        }
+      }
+    }
+  }
+}
+--------------------------------------------------
+
+<1> The `metrics` field cannot hold other objects.
+<2> Sample document holding flat paths
+<3> Sample document holding an object (configured to not hold subobjects) and its leaf sub-fields
+<4> The resulting mapping where dots in field names were preserved
+
+The entire mapping may be configured to not support subobjects as well, in which case the document can
+only ever hold leaf sub-fields:
+
+[source,console]
+--------------------------------------------------
+PUT my-index-000001
+{
+  "mappings": {
+    "subobjects": false <1>
+  }
+}
+
+PUT my-index-000001/_doc/metric_1
+{
+  "time" : "100ms", <2>
+  "time.min" : "10ms",
+  "time.max" : "900ms"
+}
+
+--------------------------------------------------
+
+<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.

docs/reference/mapping/types/object.asciidoc

@@ -90,6 +90,12 @@ The following parameters are accepted by `object` fields:
     Whether the JSON value given for the object field should be
     parsed and indexed (`true`, default) or completely ignored (`false`).
 
+<<subobjects,`subobjects`>>::
+
+    Whether the object can hold subobjects (`true`, default) or not (`false`). If not, sub-fields
+    with dots in their names will be treated as leaves instead, otherwise their field names
+    would be expanded to their corresponding object structure.
+
 <<properties,`properties`>>::
 
     The fields within the object, which can be of any

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/index/91_metrics_no_subobjects.yml

@@ -0,0 +1,37 @@
+---
+"Metrics indexing":
+  - skip:
+      version: " - 8.2.99"
+      reason: added in 8.3.0
+
+  - do:
+      indices.put_template:
+        name: test
+        body:
+          index_patterns: test-*
+          mappings:
+            dynamic_templates:
+              - no_subobjects:
+                  match: metrics
+                  mapping:
+                    type: object
+                    subobjects: false
+
+  - do:
+      index:
+        index: test-1
+        id: 1
+        refresh: true
+        body:
+          { metrics.time: 10, metrics.time.max: 100, metrics.time.min: 1 }
+
+  - do:
+      field_caps:
+        index: test-1
+        fields: metrics.time*
+  - match: {fields.metrics\.time.long.searchable: true}
+  - match: {fields.metrics\.time.long.aggregatable: true}
+  - match: {fields.metrics\.time\.max.long.searchable: true}
+  - match: {fields.metrics\.time\.max.long.aggregatable: true}
+  - match: {fields.metrics\.time\.min.long.searchable: true}
+  - match: {fields.metrics\.time\.min.long.aggregatable: true}

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

@@ -20,6 +20,8 @@ public final class ContentPath {
 
     private String[] path = new String[10];
 
+    private boolean withinLeafObject = false;
+
     public ContentPath() {
         this(0);
     }
@@ -54,6 +56,14 @@ public final class ContentPath {
         path[index--] = null;
     }
 
+    public void setWithinLeafObject(boolean withinLeafObject) {
+        this.withinLeafObject = withinLeafObject;
+    }
+
+    public boolean isWithinLeafObject() {
+        return withinLeafObject;
+    }
+
     public String pathAsText(String name) {
         sb.setLength(0);
         for (int i = offset; i < index; i++) {

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

@@ -445,7 +445,13 @@ public final class DocumentParser {
         Mapper objectMapper = getMapper(context, mapper, currentFieldName);
         if (objectMapper != null) {
             context.path().add(currentFieldName);
+            if (objectMapper instanceof ObjectMapper objMapper) {
+                if (objMapper.subobjects() == false) {
+                    context.path().setWithinLeafObject(true);
+                }
+            }
             parseObjectOrField(context, objectMapper);
+            context.path().setWithinLeafObject(false);
             context.path().remove();
         } else {
             parseObjectDynamic(context, mapper, currentFieldName);
@@ -474,7 +480,13 @@ public final class DocumentParser {
                 throwOnCreateDynamicNestedViaCopyTo(dynamicObjectMapper);
             }
             context.path().add(currentFieldName);
+            if (dynamicObjectMapper instanceof ObjectMapper objectMapper) {
+                if (objectMapper.subobjects() == false) {
+                    context.path().setWithinLeafObject(true);
+                }
+            }
             parseObjectOrField(context, dynamicObjectMapper);
+            context.path().setWithinLeafObject(false);
             context.path().remove();
         }
     }
@@ -789,7 +801,7 @@ public final class DocumentParser {
 
     private static class NoOpObjectMapper extends ObjectMapper {
         NoOpObjectMapper(String name, String fullPath) {
-            super(name, fullPath, Explicit.IMPLICIT_TRUE, Dynamic.RUNTIME, Collections.emptyMap());
+            super(name, fullPath, Explicit.IMPLICIT_TRUE, Explicit.IMPLICIT_TRUE, Dynamic.RUNTIME, Collections.emptyMap());
         }
     }
 
@@ -815,7 +827,11 @@ public final class DocumentParser {
             XContentParser parser
         ) throws IOException {
             super(mappingLookup, indexSettings, indexAnalyzers, parserContext, source);
-            this.parser = DotExpandingXContentParser.expandDots(parser);
+            if (mappingLookup.getMapping().getRoot().subobjects()) {
+                this.parser = DotExpandingXContentParser.expandDots(parser, this.path::isWithinLeafObject);
+            } else {
+                this.parser = parser;
+            }
             this.document = new LuceneDocument();
             this.documents.add(document);
             this.maxAllowedNumNestedDocs = indexSettings().getMappingNestedDocsLimit();

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

@@ -314,7 +314,7 @@ public abstract class DocumentParserContext {
      */
     public final DocumentParserContext createCopyToContext(String copyToField, LuceneDocument doc) throws IOException {
         ContentPath path = new ContentPath(0);
-        XContentParser parser = DotExpandingXContentParser.expandDots(new CopyToParser(copyToField, parser()));
+        XContentParser parser = DotExpandingXContentParser.expandDots(new CopyToParser(copyToField, parser()), path::isWithinLeafObject);
         return new Wrapper(this) {
             @Override
             public ContentPath path() {

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

@@ -20,6 +20,7 @@ import java.util.ArrayDeque;
 import java.util.Deque;
 import java.util.List;
 import java.util.Map;
+import java.util.function.BooleanSupplier;
 import java.util.function.Supplier;
 
 /**
@@ -35,9 +36,11 @@ class DotExpandingXContentParser extends FilterXContentParserWrapper {
 
     private static final class WrappingParser extends FilterXContentParser {
 
+        private final BooleanSupplier isWithinLeafObject;
         final Deque<XContentParser> parsers = new ArrayDeque<>();
 
-        WrappingParser(XContentParser in) throws IOException {
+        WrappingParser(XContentParser in, BooleanSupplier isWithinLeafObject) throws IOException {
+            this.isWithinLeafObject = isWithinLeafObject;
             parsers.push(in);
             if (in.currentToken() == Token.FIELD_NAME) {
                 expandDots();
@@ -61,6 +64,12 @@ class DotExpandingXContentParser extends FilterXContentParserWrapper {
         }
 
         private void expandDots() throws IOException {
+            // this handles fields that belong to objects that can't hold subobjects, where the document specifies
+            // the object holding the flat fields
+            // e.g. { "metrics.service": { "time.max" : 10 } } with service having subobjects set to false
+            if (isWithinLeafObject.getAsBoolean()) {
+                return;
+            }
             XContentParser delegate = delegate();
             String field = delegate.currentName();
             String[] subpaths = splitAndValidatePath(field);
@@ -82,7 +91,7 @@ class DotExpandingXContentParser extends FilterXContentParserWrapper {
                 XContentParser subParser = token == Token.START_OBJECT || token == Token.START_ARRAY
                     ? new XContentSubParser(delegate)
                     : new SingletonValueXContentParser(delegate);
-                parsers.push(new DotExpandingXContentParser(subParser, subpaths, location));
+                parsers.push(new DotExpandingXContentParser(subParser, subpaths, location, isWithinLeafObject));
             }
         }
 
@@ -123,25 +132,26 @@ class DotExpandingXContentParser extends FilterXContentParserWrapper {
         }
     }
 
-    private static String[] splitAndValidatePath(String fullFieldPath) {
-        if (fullFieldPath.isEmpty()) {
+    private static String[] splitAndValidatePath(String fieldName) {
+        if (fieldName.isEmpty()) {
             throw new IllegalArgumentException("field name cannot be an empty string");
         }
-        if (fullFieldPath.contains(".") == false) {
-            return new String[] { fullFieldPath };
+        if (fieldName.contains(".") == false) {
+            return new String[] { fieldName };
         }
-        String[] parts = fullFieldPath.split("\\.");
+        String[] parts = fieldName.split("\\.");
         if (parts.length == 0) {
             throw new IllegalArgumentException("field name cannot contain only dots");
         }
+
         for (String part : parts) {
             // check if the field name contains only whitespace
             if (part.isEmpty()) {
-                throw new IllegalArgumentException("field name cannot contain only whitespace: ['" + fullFieldPath + "']");
+                throw new IllegalArgumentException("field name cannot contain only whitespace: ['" + fieldName + "']");
             }
             if (part.isBlank()) {
                 throw new IllegalArgumentException(
-                    "field name starting or ending with a [.] makes object resolution ambiguous: [" + fullFieldPath + "]"
+                    "field name starting or ending with a [.] makes object resolution ambiguous: [" + fieldName + "]"
                 );
             }
         }
@@ -153,8 +163,8 @@ class DotExpandingXContentParser extends FilterXContentParserWrapper {
      * @param in    the parser to wrap
      * @return  the wrapped XContentParser
      */
-    static XContentParser expandDots(XContentParser in) throws IOException {
-        return new WrappingParser(in);
+    static XContentParser expandDots(XContentParser in, BooleanSupplier isWithinLeafObject) throws IOException {
+        return new WrappingParser(in, isWithinLeafObject);
     }
 
     private enum State {
@@ -163,17 +173,24 @@ class DotExpandingXContentParser extends FilterXContentParserWrapper {
         ENDING_EXPANDED_OBJECT
     }
 
-    final String[] subPaths;
+    private final BooleanSupplier isWithinLeafObject;
 
+    private String[] subPaths;
     private XContentLocation currentLocation;
     private int expandedTokens = 0;
     private int innerLevel = -1;
     private State state = State.EXPANDING_START_OBJECT;
 
-    private DotExpandingXContentParser(XContentParser subparser, String[] subPaths, XContentLocation startLocation) {
+    private DotExpandingXContentParser(
+        XContentParser subparser,
+        String[] subPaths,
+        XContentLocation startLocation,
+        BooleanSupplier isWithinLeafObject
+    ) {
         super(subparser);
         this.subPaths = subPaths;
         this.currentLocation = startLocation;
+        this.isWithinLeafObject = isWithinLeafObject;
     }
 
     @Override
@@ -191,6 +208,25 @@ class DotExpandingXContentParser extends FilterXContentParserWrapper {
             }
             // The expansion consists of adding pairs of START_OBJECT and FIELD_NAME tokens
             if (expandedTokens % 2 == 0) {
+                int currentIndex = expandedTokens / 2;
+                // if there's more than one element left to expand and the parent can't hold subobjects, we replace the array
+                // e.g. metrics.service.time.max -> ["metrics", "service", "time.max"]
+                if (currentIndex < subPaths.length - 1 && isWithinLeafObject.getAsBoolean()) {
+                    String[] newSubPaths = new String[currentIndex + 1];
+                    StringBuilder collapsedPath = new StringBuilder();
+                    for (int i = 0; i < subPaths.length; i++) {
+                        if (i < currentIndex) {
+                            newSubPaths[i] = subPaths[i];
+                        } else {
+                            collapsedPath.append(subPaths[i]);
+                            if (i < subPaths.length - 1) {
+                                collapsedPath.append(".");
+                            }
+                        }
+                    }
+                    newSubPaths[currentIndex] = collapsedPath.toString();
+                    subPaths = newSubPaths;
+                }
                 return Token.FIELD_NAME;
             }
             return Token.START_OBJECT;

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

@@ -67,6 +67,9 @@ public class NestedObjectMapper extends ObjectMapper {
                     iterator.remove();
                 }
             }
+            if (builder.subobjects.explicit()) {
+                throw new MapperParsingException("Nested type [" + name + "] does not support [subobjects] parameter");
+            }
             return builder;
         }
 
@@ -92,7 +95,7 @@ public class NestedObjectMapper extends ObjectMapper {
     private final Query nestedTypeFilter;
 
     NestedObjectMapper(String name, String fullPath, Map<String, Mapper> mappers, Builder builder) {
-        super(name, fullPath, builder.enabled, builder.dynamic, mappers);
+        super(name, fullPath, builder.enabled, Explicit.IMPLICIT_TRUE, builder.dynamic, mappers);
         if (builder.indexCreatedVersion.before(Version.V_8_0_0)) {
             this.nestedTypePath = "__" + fullPath;
         } else {

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

@@ -63,9 +63,8 @@ public class ObjectMapper extends Mapper implements Cloneable {
     public static class Builder extends Mapper.Builder {
 
         protected Explicit<Boolean> enabled = Explicit.IMPLICIT_TRUE;
-
+        protected Explicit<Boolean> subobjects = Explicit.IMPLICIT_TRUE;
         protected Dynamic dynamic;
-
         protected final List<Mapper.Builder> mappersBuilders = new ArrayList<>();
 
         public Builder(String name) {
@@ -77,6 +76,11 @@ public class ObjectMapper extends Mapper implements Cloneable {
             return this;
         }
 
+        public Builder subobjects(boolean subobjects) {
+            this.subobjects = Explicit.explicitBoolean(subobjects);
+            return this;
+        }
+
         public Builder dynamic(Dynamic dynamic) {
             this.dynamic = dynamic;
             return this;
@@ -87,13 +91,17 @@ public class ObjectMapper extends Mapper implements Cloneable {
             return this;
         }
 
-        Builder addMappers(Map<String, Mapper> mappers) {
-            mappers.forEach((name, mapper) -> mappersBuilders.add(new Mapper.Builder(name) {
+        private void add(String name, Mapper mapper) {
+            add(new Mapper.Builder(name) {
                 @Override
                 public Mapper build(MapperBuilderContext context) {
                     return mapper;
                 }
-            }));
+            });
+        }
+
+        Builder addMappers(Map<String, Mapper> mappers) {
+            mappers.forEach(this::add);
             return this;
         }
 
@@ -106,15 +114,10 @@ public class ObjectMapper extends Mapper implements Cloneable {
          * @param context   the DocumentParserContext in which the mapper has been built
          */
         public void addDynamic(String name, String prefix, Mapper mapper, DocumentParserContext context) {
-            // If the mapper to add has no dots and is therefore
-            // a leaf mapper, we just add it here
-            if (name.contains(".") == false) {
-                mappersBuilders.add(new Mapper.Builder(name) {
-                    @Override
-                    public Mapper build(MapperBuilderContext context) {
-                        return 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.value() == false) {
+                add(name, mapper);
             }
             // otherwise 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
@@ -126,7 +129,7 @@ public class ObjectMapper extends Mapper implements Cloneable {
                 String fullChildName = prefix == null ? childName : prefix + "." + childName;
                 ObjectMapper.Builder childBuilder = findChild(fullChildName, context);
                 childBuilder.addDynamic(name.substring(firstDotIndex + 1), fullChildName, mapper, context);
-                mappersBuilders.add(childBuilder);
+                add(childBuilder);
             }
         }
 
@@ -145,12 +148,19 @@ public class ObjectMapper extends Mapper implements Cloneable {
         }
 
         protected final Map<String, Mapper> buildMappers(boolean root, MapperBuilderContext context) {
-            if (root == false) {
-                context = context.createChildContext(name);
-            }
+            MapperBuilderContext mapperBuilderContext = root ? context : context.createChildContext(name);
             Map<String, Mapper> mappers = new HashMap<>();
             for (Mapper.Builder builder : mappersBuilders) {
-                Mapper mapper = builder.build(context);
+                Mapper mapper = builder.build(mapperBuilderContext);
+                if (subobjects.value() == false && mapper instanceof ObjectMapper) {
+                    throw new IllegalArgumentException(
+                        "Object ["
+                            + context.buildFullName(name)
+                            + "] has subobjects set to false hence it does not support inner object ["
+                            + mapper.simpleName()
+                            + "]"
+                    );
+                }
                 Mapper existing = mappers.get(mapper.simpleName());
                 if (existing != null) {
                     mapper = existing.merge(mapper);
@@ -162,7 +172,7 @@ public class ObjectMapper extends Mapper implements Cloneable {
 
         @Override
         public ObjectMapper build(MapperBuilderContext context) {
-            return new ObjectMapper(name, context.buildFullName(name), enabled, dynamic, buildMappers(false, context));
+            return new ObjectMapper(name, context.buildFullName(name), enabled, subobjects, dynamic, buildMappers(false, context));
         }
     }
 
@@ -209,6 +219,9 @@ public class ObjectMapper extends Mapper implements Cloneable {
             } else if (fieldName.equals("enabled")) {
                 builder.enabled(XContentMapValues.nodeBooleanValue(fieldNode, fieldName + ".enabled"));
                 return true;
+            } else if (fieldName.equals("subobjects")) {
+                builder.subobjects(XContentMapValues.nodeBooleanValue(fieldNode, fieldName + ".subobjects"));
+                return true;
             } else if (fieldName.equals("properties")) {
                 if (fieldNode instanceof Collection && ((Collection) fieldNode).isEmpty()) {
                     // nothing to do here, empty (to support "properties: []" case)
@@ -264,17 +277,31 @@ public class ObjectMapper extends Mapper implements Cloneable {
                         }
                     }
 
+                    if (objBuilder.subobjects.value() == false && type.equals(ObjectMapper.CONTENT_TYPE)) {
+                        throw new MapperException(
+                            "Object ["
+                                + objBuilder.name()
+                                + "] has subobjects set to false hence it does not support inner object ["
+                                + fieldName
+                                + "]"
+                        );
+                    }
                     Mapper.TypeParser typeParser = parserContext.typeParser(type);
                     if (typeParser == null) {
                         throw new MapperParsingException("No handler for type [" + type + "] declared on field [" + fieldName + "]");
                     }
-                    String[] fieldNameParts = fieldName.split("\\.");
-                    String realFieldName = fieldNameParts[fieldNameParts.length - 1];
-                    Mapper.Builder fieldBuilder = typeParser.parse(realFieldName, propNode, parserContext);
-                    for (int i = fieldNameParts.length - 2; i >= 0; --i) {
-                        ObjectMapper.Builder intermediate = new ObjectMapper.Builder(fieldNameParts[i]);
-                        intermediate.add(fieldBuilder);
-                        fieldBuilder = intermediate;
+                    Mapper.Builder fieldBuilder;
+                    if (objBuilder.subobjects.value() == false) {
+                        fieldBuilder = typeParser.parse(fieldName, propNode, parserContext);
+                    } else {
+                        String[] fieldNameParts = fieldName.split("\\.");
+                        String realFieldName = fieldNameParts[fieldNameParts.length - 1];
+                        fieldBuilder = typeParser.parse(realFieldName, propNode, parserContext);
+                        for (int i = fieldNameParts.length - 2; i >= 0; --i) {
+                            ObjectMapper.Builder intermediate = new ObjectMapper.Builder(fieldNameParts[i]);
+                            intermediate.add(fieldBuilder);
+                            fieldBuilder = intermediate;
+                        }
                     }
                     objBuilder.add(fieldBuilder);
                     propNode.remove("type");
@@ -296,17 +323,26 @@ public class ObjectMapper extends Mapper implements Cloneable {
     private final String fullPath;
 
     protected Explicit<Boolean> enabled;
+    protected Explicit<Boolean> subobjects;
     protected volatile Dynamic dynamic;
 
     protected Map<String, Mapper> mappers;
 
-    ObjectMapper(String name, String fullPath, Explicit<Boolean> enabled, Dynamic dynamic, Map<String, Mapper> mappers) {
+    ObjectMapper(
+        String name,
+        String fullPath,
+        Explicit<Boolean> enabled,
+        Explicit<Boolean> subobjects,
+        Dynamic dynamic,
+        Map<String, Mapper> mappers
+    ) {
         super(name);
         if (name.isEmpty()) {
             throw new IllegalArgumentException("name cannot be empty string");
         }
         this.fullPath = internFieldName(fullPath);
         this.enabled = enabled;
+        this.subobjects = subobjects;
         this.dynamic = dynamic;
         if (mappers == null) {
             this.mappers = Map.of();
@@ -333,6 +369,7 @@ public class ObjectMapper extends Mapper implements Cloneable {
     public ObjectMapper.Builder newBuilder(Version indexVersionCreated) {
         ObjectMapper.Builder builder = new ObjectMapper.Builder(simpleName());
         builder.enabled = this.enabled;
+        builder.subobjects = this.subobjects;
         builder.dynamic = this.dynamic;
         return builder;
     }
@@ -372,6 +409,10 @@ public class ObjectMapper extends Mapper implements Cloneable {
         return dynamic;
     }
 
+    public final boolean subobjects() {
+        return subobjects.value();
+    }
+
     @Override
     public ObjectMapper merge(Mapper mergeWith) {
         return merge(mergeWith, MergeReason.MAPPING_UPDATE);
@@ -412,6 +453,14 @@ public class ObjectMapper extends Mapper implements Cloneable {
             }
         }
 
+        if (mergeWith.subobjects.explicit()) {
+            if (reason == MergeReason.INDEX_TEMPLATE) {
+                this.subobjects = mergeWith.subobjects;
+            } else if (subobjects != mergeWith.subobjects) {
+                throw new MapperException("the [subobjects] parameter can't be updated for the object mapping [" + name() + "]");
+            }
+        }
+
         Map<String, Mapper> mergedMappers = null;
         for (Mapper mergeWithMapper : mergeWith) {
             Mapper mergeIntoMapper = (mergedMappers == null ? mappers : mergedMappers).get(mergeWithMapper.simpleName());
@@ -465,6 +514,9 @@ public class ObjectMapper extends Mapper implements Cloneable {
         if (isEnabled() != Defaults.ENABLED) {
             builder.field("enabled", enabled.value());
         }
+        if (subobjects() == false) {
+            builder.field("subobjects", subobjects.value());
+        }
         if (custom != null) {
             custom.toXContent(builder, params);
         }

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

@@ -105,6 +105,7 @@ public class RootObjectMapper extends ObjectMapper {
             return new RootObjectMapper(
                 name,
                 enabled,
+                subobjects,
                 dynamic,
                 buildMappers(true, context),
                 runtimeFields,
@@ -251,6 +252,7 @@ public class RootObjectMapper extends ObjectMapper {
     RootObjectMapper(
         String name,
         Explicit<Boolean> enabled,
+        Explicit<Boolean> subobjects,
         Dynamic dynamic,
         Map<String, Mapper> mappers,
         Map<String, RuntimeField> runtimeFields,
@@ -259,7 +261,7 @@ public class RootObjectMapper extends ObjectMapper {
         Explicit<Boolean> dateDetection,
         Explicit<Boolean> numericDetection
     ) {
-        super(name, name, enabled, dynamic, mappers);
+        super(name, name, enabled, subobjects, dynamic, mappers);
         this.runtimeFields = runtimeFields;
         this.dynamicTemplates = dynamicTemplates;
         this.dynamicDateTimeFormatters = dynamicDateTimeFormatters;
@@ -278,6 +280,7 @@ public class RootObjectMapper extends ObjectMapper {
     public RootObjectMapper.Builder newBuilder(Version indexVersionCreated) {
         RootObjectMapper.Builder builder = new RootObjectMapper.Builder(name());
         builder.enabled = enabled;
+        builder.subobjects = subobjects;
         builder.dynamic = dynamic;
         return builder;
     }

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

@@ -1842,6 +1842,196 @@ public class DocumentParserTests extends MapperServiceTestCase {
         assertThat(err.getCause().getMessage(), containsString("field name cannot contain only dots"));
     }
 
+    public void testSubobjectsFalseWithInnerObject() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            mapping(b -> b.startObject("metrics.service").field("type", "object").field("subobjects", false).endObject())
+        );
+        IllegalArgumentException err = expectThrows(IllegalArgumentException.class, () -> mapper.parse(source("""
+            {
+              "metrics": {
+                "service": {
+                  "time" : {
+                    "max" : 10
+                  }
+                }
+              }
+            }
+            """)));
+        assertEquals(
+            "Object [metrics.service] has subobjects set to false hence it does not support inner object [time]",
+            err.getMessage()
+        );
+    }
+
+    public void testSubobjectsFalseWithInnerDottedObject() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            mapping(b -> b.startObject("metrics.service").field("type", "object").field("subobjects", false).endObject())
+        );
+        IllegalArgumentException err = expectThrows(IllegalArgumentException.class, () -> mapper.parse(source("""
+            {
+              "metrics": {
+                "service": {
+                  "test.with.dots" : {
+                    "max" : 10
+                  }
+                }
+              }
+            }
+            """)));
+        assertEquals(
+            "Object [metrics.service] has subobjects set to false hence it does not support inner object [test.with.dots]",
+            err.getMessage()
+        );
+    }
+
+    public void testSubobjectsFalseRootWithInnerObject() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(topMapping(b -> b.field("subobjects", false)));
+        IllegalArgumentException err = expectThrows(IllegalArgumentException.class, () -> mapper.parse(source("""
+            {
+              "metrics": {
+                "service": {
+                  "time.max" : 10
+                }
+              }
+            }
+            """)));
+        assertEquals("Object [_doc] has subobjects set to false hence it does not support inner object [metrics]", err.getMessage());
+    }
+
+    public void testSubobjectsFalseRoot() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(topMapping(b -> b.field("subobjects", false)));
+        ParsedDocument doc = mapper.parse(source("""
+            {
+              "metrics.service.time" : 10,
+              "metrics.service.time.max" : 500,
+              "metrics.service.time.min" : 1,
+              "metrics.object.inner.field" : 1,
+              "metrics.service.test.with.dots" : "value"
+            }
+            """));
+
+        Mapping mappingsUpdate = doc.dynamicMappingsUpdate();
+        assertNotNull(mappingsUpdate);
+        assertNotNull(mappingsUpdate.getRoot().getMapper("metrics.service.time"));
+        assertNotNull(mappingsUpdate.getRoot().getMapper("metrics.service.time.max"));
+        assertNotNull(mappingsUpdate.getRoot().getMapper("metrics.service.test.with.dots"));
+
+        assertNotNull(doc.rootDoc().getFields("metrics.service.time"));
+        assertNotNull(doc.rootDoc().getFields("metrics.service.time.max"));
+        assertNotNull(doc.rootDoc().getFields("metrics.service.test.with.dots"));
+    }
+
+    public void testSubobjectsFalseStructuredPath() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            mapping(b -> b.startObject("metrics.service").field("type", "object").field("subobjects", false).endObject())
+        );
+        ParsedDocument doc = mapper.parse(source("""
+            {
+              "metrics": {
+                "service": {
+                  "time" : 10,
+                  "time.max" : 500,
+                  "time.min" : 1,
+                  "test.with.dots" : "value"
+                },
+                "object.inner.field": "value"
+              }
+            }
+            """));
+        assertNoSubobjects(doc);
+    }
+
+    public void testSubobjectsFalseFlatPaths() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            mapping(b -> b.startObject("metrics.service").field("type", "object").field("subobjects", false).endObject())
+        );
+        ParsedDocument doc = mapper.parse(source("""
+            {
+              "metrics.service.time" : 10,
+              "metrics.service.time.max" : 500,
+              "metrics.service.time.min" : 1,
+              "metrics.service.test.with.dots" : "value",
+              "metrics.object.inner.field" : "value"
+            }
+            """));
+        assertNoSubobjects(doc);
+    }
+
+    public void testSubobjectsFalseMixedPaths() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            mapping(b -> b.startObject("metrics.service").field("type", "object").field("subobjects", false).endObject())
+        );
+        ParsedDocument doc = mapper.parse(source("""
+            {
+              "metrics": {
+                "service.time": 10,
+                "service": {
+                  "time.max" : 500,
+                  "time.min" : 1
+                },
+                "object.inner.field" : "value"
+              },
+              "metrics.service.test.with.dots" : "value"
+            }
+            """));
+        assertNoSubobjects(doc);
+    }
+
+    public void testSubobjectsFalseArrayOfObjects() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            mapping(b -> b.startObject("metrics.service").field("type", "object").field("subobjects", false).endObject())
+        );
+        ParsedDocument doc = mapper.parse(source("""
+            {
+              "metrics": {
+                "service": [
+                  {
+                    "time" : 10,
+                    "time.max" : 500,
+                    "time.min" : 1,
+                    "test.with.dots" : "value1"
+                  },
+                  {
+                    "time" : 5,
+                    "time.max" : 600,
+                    "time.min" : 3,
+                    "test.with.dots" : "value2"
+                  }
+                ],
+                "object.inner.field": "value"
+              }
+            }
+            """));
+        assertNoSubobjects(doc);
+    }
+
+    private static void assertNoSubobjects(ParsedDocument doc) {
+        Mapping mappingsUpdate = doc.dynamicMappingsUpdate();
+        assertNotNull(mappingsUpdate);
+        Mapper metrics = mappingsUpdate.getRoot().mappers.get("metrics");
+        assertThat(metrics, instanceOf(ObjectMapper.class));
+        ObjectMapper metricsObject = (ObjectMapper) metrics;
+        Mapper service = metricsObject.getMapper("service");
+        assertThat(service, instanceOf(ObjectMapper.class));
+        ObjectMapper serviceObject = (ObjectMapper) service;
+        assertNotNull(serviceObject.getMapper("time"));
+        assertNotNull(serviceObject.getMapper("time.max"));
+        assertNotNull(serviceObject.getMapper("time.min"));
+        assertNotNull(serviceObject.getMapper("test.with.dots"));
+        Mapper object = metricsObject.getMapper("object");
+        assertThat(object, instanceOf(ObjectMapper.class));
+        ObjectMapper objectObject = (ObjectMapper) object;
+        Mapper inner = objectObject.getMapper("inner");
+        assertThat(inner, instanceOf(ObjectMapper.class));
+        ObjectMapper innerObject = (ObjectMapper) inner;
+        assertNotNull(innerObject.getMapper("field"));
+
+        assertNotNull(doc.rootDoc().getFields("metrics.service.time"));
+        assertNotNull(doc.rootDoc().getFields("metrics.service.time.max"));
+        assertNotNull(doc.rootDoc().getFields("metrics.service.time.min"));
+        assertNotNull(doc.rootDoc().getFields("metrics.service.test.with.dots"));
+    }
+
     public void testWriteToFieldAlias() throws Exception {
         DocumentMapper mapper = createDocumentMapper(mapping(b -> {
             b.startObject("alias-field");

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

@@ -20,7 +20,7 @@ public class DotExpandingXContentParserTests extends ESTestCase {
 
     private void assertXContentMatches(String dotsExpanded, String withDots) throws IOException {
         XContentParser inputParser = createParser(JsonXContent.jsonXContent, withDots);
-        XContentParser expandedParser = DotExpandingXContentParser.expandDots(inputParser);
+        XContentParser expandedParser = DotExpandingXContentParser.expandDots(inputParser, () -> false);
         expandedParser.allowDuplicateKeys(true);
 
         XContentBuilder actualOutput = XContentBuilder.builder(JsonXContent.jsonXContent).copyCurrentStructure(expandedParser);
@@ -28,13 +28,12 @@ public class DotExpandingXContentParserTests extends ESTestCase {
 
         XContentParser expectedParser = createParser(JsonXContent.jsonXContent, dotsExpanded);
         expectedParser.allowDuplicateKeys(true);
-        XContentParser actualParser = DotExpandingXContentParser.expandDots(createParser(JsonXContent.jsonXContent, withDots));
+        XContentParser actualParser = DotExpandingXContentParser.expandDots(createParser(JsonXContent.jsonXContent, withDots), () -> false);
         XContentParser.Token currentToken;
         while ((currentToken = actualParser.nextToken()) != null) {
             assertEquals(currentToken, expectedParser.nextToken());
             assertEquals(expectedParser.currentToken(), actualParser.currentToken());
             assertEquals(actualParser.currentToken().name(), expectedParser.currentName(), actualParser.currentName());
-
         }
         assertNull(expectedParser.nextToken());
     }
@@ -112,10 +111,118 @@ public class DotExpandingXContentParserTests extends ESTestCase {
               "test.with.dots2" : "value2"}""");
     }
 
-    public void testSkipChildren() throws IOException {
+    public void testDotsCollapsingFlatPaths() throws IOException {
+        ContentPath contentPath = new ContentPath();
         XContentParser parser = DotExpandingXContentParser.expandDots(createParser(JsonXContent.jsonXContent, """
-            { "test.with.dots" : "value", "nodots" : "value2" }"""));
+            {"metrics.service.time": 10, "metrics.service.time.max": 500, "metrics.foo": "value"}"""), contentPath::isWithinLeafObject);
+        parser.nextToken();
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("metrics", parser.currentName());
+        assertEquals(XContentParser.Token.START_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.START_OBJECT, parser.currentToken());
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("service", parser.currentName());
+        contentPath.setWithinLeafObject(true);
+        assertEquals(XContentParser.Token.START_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.START_OBJECT, parser.currentToken());
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("time", parser.currentName());
+        assertEquals(XContentParser.Token.VALUE_NUMBER, parser.nextToken());
+        assertEquals("time", parser.currentName());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.currentToken());
+        assertEquals("service", parser.currentName());
+        contentPath.setWithinLeafObject(false);
+        assertEquals(XContentParser.Token.END_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.currentToken());
+        assertEquals("metrics", parser.currentName());
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("metrics", parser.currentName());
+        assertEquals(XContentParser.Token.START_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.START_OBJECT, parser.currentToken());
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("service", parser.currentName());
+        contentPath.setWithinLeafObject(true);
+        assertEquals(XContentParser.Token.START_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.START_OBJECT, parser.currentToken());
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("time.max", parser.currentName());
+        assertEquals(XContentParser.Token.VALUE_NUMBER, parser.nextToken());
+        assertEquals("time.max", parser.currentName());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.currentToken());
+        assertEquals("service", parser.currentName());
+        contentPath.setWithinLeafObject(false);
+        assertEquals(XContentParser.Token.END_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.currentToken());
+        assertEquals("metrics", parser.currentName());
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("metrics", parser.currentName());
+        assertEquals(XContentParser.Token.START_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.START_OBJECT, parser.currentToken());
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("foo", parser.currentName());
+        assertEquals(XContentParser.Token.VALUE_STRING, parser.nextToken());
+        assertEquals("foo", parser.currentName());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.currentToken());
+        assertEquals("metrics", parser.currentName());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.currentToken());
+        assertNull(parser.currentName());
+        assertNull(parser.nextToken());
+    }
+
+    public void testDotsCollapsingStructuredPath() throws IOException {
+        ContentPath contentPath = new ContentPath();
+        XContentParser parser = DotExpandingXContentParser.expandDots(createParser(JsonXContent.jsonXContent, """
+            {
+              "metrics" : {
+                "service" : {
+                  "time" : 10,
+                  "time.max" : 500
+                },
+                "foo" : "value"
+              }
+            }"""), contentPath::isWithinLeafObject);
+        parser.nextToken();
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("metrics", parser.currentName());
+        assertEquals(XContentParser.Token.START_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.START_OBJECT, parser.currentToken());
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("service", parser.currentName());
+        contentPath.setWithinLeafObject(true);
+        assertEquals(XContentParser.Token.START_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.START_OBJECT, parser.currentToken());
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("time", parser.currentName());
+        assertEquals(XContentParser.Token.VALUE_NUMBER, parser.nextToken());
+        assertEquals("time", parser.currentName());
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("time.max", parser.currentName());
+        assertEquals(XContentParser.Token.VALUE_NUMBER, parser.nextToken());
+        assertEquals("time.max", parser.currentName());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.currentToken());
+        assertEquals("service", parser.currentName());
+        contentPath.setWithinLeafObject(false);
+        assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+        assertEquals("foo", parser.currentName());
+        assertEquals(XContentParser.Token.VALUE_STRING, parser.nextToken());
+        assertEquals("foo", parser.currentName());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.currentToken());
+        assertEquals("metrics", parser.currentName());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.nextToken());
+        assertEquals(XContentParser.Token.END_OBJECT, parser.currentToken());
+        assertNull(parser.currentName());
+        assertNull(parser.nextToken());
+    }
 
+    public void testSkipChildren() throws IOException {
+        XContentParser parser = DotExpandingXContentParser.expandDots(createParser(JsonXContent.jsonXContent, """
+            { "test.with.dots" : "value", "nodots" : "value2" }"""), () -> false);
         parser.nextToken();     // start object
         assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
         assertEquals("test", parser.currentName());
@@ -138,7 +245,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" }"""));
+            { "test.with.dots" : {"obj" : {"field":"value"}}, "nodots" : "value2" }"""), () -> false);
 
         parser.nextToken();     // start object
         assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
@@ -184,7 +291,10 @@ public class DotExpandingXContentParserTests extends ESTestCase {
             "value":null}}\
             """;
         XContentParser expectedParser = createParser(JsonXContent.jsonXContent, jsonInput);
-        XContentParser dotExpandedParser = DotExpandingXContentParser.expandDots(createParser(JsonXContent.jsonXContent, jsonInput));
+        XContentParser dotExpandedParser = DotExpandingXContentParser.expandDots(
+            createParser(JsonXContent.jsonXContent, jsonInput),
+            () -> false
+        );
 
         assertEquals(expectedParser.getTokenLocation(), dotExpandedParser.getTokenLocation());
         assertEquals(XContentParser.Token.START_OBJECT, dotExpandedParser.nextToken());

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

@@ -1077,4 +1077,101 @@ public class DynamicTemplatesTests extends MapperServiceTestCase {
               }
             }"""), Strings.toString(parsedDoc2.dynamicMappingsUpdate()));
     }
+
+    private MapperService createDynamicTemplateNoSubobjects() throws IOException {
+        return createMapperService(topMapping(b -> {
+            b.startArray("dynamic_templates");
+            {
+                b.startObject();
+                {
+                    b.startObject("test");
+                    {
+                        b.field("match_mapping_type", "object");
+                        b.field("match", "metric");
+                        b.startObject("mapping").field("type", "object").field("subobjects", false).endObject();
+                    }
+                    b.endObject();
+                }
+                b.endObject();
+            }
+            b.endArray();
+        }));
+    }
+
+    private static void assertNoSubobjects(MapperService mapperService) {
+        assertThat(mapperService.fieldType("foo.bar.baz").typeName(), equalTo("long"));
+        assertNotNull(mapperService.mappingLookup().objectMappers().get("foo.bar"));
+        assertThat(mapperService.fieldType("foo.metric.count").typeName(), equalTo("long"));
+        assertThat(mapperService.fieldType("foo.metric.count.min").typeName(), equalTo("long"));
+        assertThat(mapperService.fieldType("foo.metric.count.max").typeName(), equalTo("long"));
+        assertNotNull(mapperService.mappingLookup().objectMappers().get("foo.metric"));
+        assertNull(mapperService.mappingLookup().objectMappers().get("foo.metric.count"));
+    }
+
+    public void testSubobjectsFalseFlatPaths() throws IOException {
+        MapperService mapperService = createDynamicTemplateNoSubobjects();
+        ParsedDocument doc = mapperService.documentMapper().parse(source(b -> {
+            b.field("foo.metric.count", 10);
+            b.field("foo.bar.baz", 10);
+            b.field("foo.metric.count.min", 4);
+            b.field("foo.metric.count.max", 15);
+        }));
+        merge(mapperService, dynamicMapping(doc.dynamicMappingsUpdate()));
+        assertNoSubobjects(mapperService);
+    }
+
+    public void testSubobjectsFalseStructuredPaths() throws IOException {
+        MapperService mapperService = createDynamicTemplateNoSubobjects();
+        ParsedDocument doc = mapperService.documentMapper().parse(source(b -> {
+            b.startObject("foo");
+            {
+                b.startObject("metric");
+                {
+                    b.field("count", 10);
+                    b.field("count.min", 4);
+                    b.field("count.max", 15);
+                }
+                b.endObject();
+                b.startObject("bar");
+                b.field("baz", 10);
+                b.endObject();
+            }
+            b.endObject();
+        }));
+        merge(mapperService, dynamicMapping(doc.dynamicMappingsUpdate()));
+        assertNoSubobjects(mapperService);
+    }
+
+    public void testSubobjectsFalseArrayOfObjects() throws IOException {
+        MapperService mapperService = createDynamicTemplateNoSubobjects();
+        ParsedDocument doc = mapperService.documentMapper().parse(source(b -> {
+            b.startObject("foo");
+            {
+                b.startArray("metric");
+                {
+                    b.startObject();
+                    {
+                        b.field("count", 10);
+                        b.field("count.min", 4);
+                        b.field("count.max", 15);
+                    }
+                    b.endObject();
+                    b.startObject();
+                    {
+                        b.field("count", 5);
+                        b.field("count.min", 3);
+                        b.field("count.max", 50);
+                    }
+                    b.endObject();
+                }
+                b.endArray();
+                b.startObject("bar");
+                b.field("baz", 10);
+                b.endObject();
+            }
+            b.endObject();
+        }));
+        merge(mapperService, dynamicMapping(doc.dynamicMappingsUpdate()));
+        assertNoSubobjects(mapperService);
+    }
 }

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

@@ -164,7 +164,7 @@ public class FieldAliasMapperValidationTests extends ESTestCase {
     }
 
     private static ObjectMapper createObjectMapper(String name) {
-        return new ObjectMapper(name, name, Explicit.IMPLICIT_TRUE, ObjectMapper.Dynamic.FALSE, emptyMap());
+        return new ObjectMapper(name, name, Explicit.IMPLICIT_TRUE, Explicit.IMPLICIT_TRUE, ObjectMapper.Dynamic.FALSE, emptyMap());
     }
 
     private static NestedObjectMapper createNestedObjectMapper(String name) {

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

@@ -77,6 +77,7 @@ public class MappingLookupTests extends ESTestCase {
             "object",
             "object",
             Explicit.EXPLICIT_TRUE,
+            Explicit.IMPLICIT_TRUE,
             ObjectMapper.Dynamic.TRUE,
             Collections.singletonMap("object.subfield", fieldMapper)
         );

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

@@ -1429,4 +1429,14 @@ public class NestedObjectMapperTests extends MapperServiceTestCase {
             assertThat(e.getMessage(), containsString("time_series_dimension can't be configured in nested field [nested.object.foo]"));
         }
     }
+
+    public void testNestedDoesNotSupportSubobjectsParameter() {
+        MapperParsingException exception = expectThrows(
+            MapperParsingException.class,
+            () -> createDocumentMapper(
+                mapping(b -> b.startObject("nested1").field("type", "nested").field("subobjects", randomBoolean()).endObject())
+            )
+        );
+        assertEquals("Failed to parse mapping: Nested type [nested1] does not support [subobjects] parameter", exception.getMessage());
+    }
 }

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

@@ -156,6 +156,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
         ObjectMapper objectMapper = mapper.mappers().objectMappers().get("object");
         assertNotNull(objectMapper);
         assertFalse(objectMapper.isEnabled());
+        assertTrue(objectMapper.subobjects());
 
         // Setting 'enabled' to true is allowed, and updates the mapping.
         update = Strings.toString(
@@ -165,6 +166,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
                 .startObject("object")
                 .field("type", "object")
                 .field("enabled", true)
+                .field("subobjects", false)
                 .endObject()
                 .endObject()
                 .endObject()
@@ -174,6 +176,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
         objectMapper = mapper.mappers().objectMappers().get("object");
         assertNotNull(objectMapper);
         assertTrue(objectMapper.isEnabled());
+        assertFalse(objectMapper.subobjects());
     }
 
     public void testFieldReplacementForIndexTemplates() throws IOException {
@@ -346,4 +349,129 @@ public class ObjectMapperTests extends MapperServiceTestCase {
         }));
         assertThat(service.fieldType("name"), instanceOf(PlaceHolderFieldMapper.PlaceHolderFieldType.class));
     }
+
+    public void testSubobjectsFalse() throws Exception {
+        MapperService mapperService = createMapperService(mapping(b -> {
+            b.startObject("metrics.service");
+            {
+                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.endObject();
+            }
+            b.endObject();
+        }));
+        assertNotNull(mapperService.fieldType("metrics.service.time"));
+        assertNotNull(mapperService.fieldType("metrics.service.time.max"));
+    }
+
+    public void testSubobjectsFalseWithInnerObject() {
+        MapperParsingException exception = expectThrows(MapperParsingException.class, () -> createMapperService(mapping(b -> {
+            b.startObject("metrics.service");
+            {
+                b.field("subobjects", false);
+                b.startObject("properties");
+                {
+                    b.startObject("time");
+                    {
+                        b.startObject("properties");
+                        {
+                            b.startObject("max");
+                            b.field("type", "long");
+                            b.endObject();
+                        }
+                        b.endObject();
+                    }
+                    b.endObject();
+                }
+                b.endObject();
+            }
+            b.endObject();
+        })));
+        assertEquals(
+            "Failed to parse mapping: Object [service] has subobjects set to false hence it does not support inner object [time]",
+            exception.getMessage()
+        );
+    }
+
+    public void testSubobjectsFalseRoot() throws Exception {
+        MapperService mapperService = createMapperService(topMapping(b -> {
+            b.field("subobjects", false);
+            b.startObject("properties");
+            {
+                b.startObject("metrics.service.time");
+                b.field("type", "long");
+                b.endObject();
+                b.startObject("metrics.service.time.max");
+                b.field("type", "long");
+                b.endObject();
+            }
+            b.endObject();
+        }));
+        assertNotNull(mapperService.fieldType("metrics.service.time"));
+        assertNotNull(mapperService.fieldType("metrics.service.time.max"));
+    }
+
+    public void testSubobjectsFalseRootWithInnerObject() {
+        MapperParsingException exception = expectThrows(MapperParsingException.class, () -> createMapperService(topMapping(b -> {
+            b.field("subobjects", false);
+            b.startObject("properties");
+            {
+                b.startObject("metrics.service.time");
+                {
+                    b.startObject("properties");
+                    {
+                        b.startObject("max");
+                        b.field("type", "long");
+                        b.endObject();
+                    }
+                    b.endObject();
+                }
+                b.endObject();
+            }
+            b.endObject();
+        })));
+        assertEquals(
+            "Failed to parse mapping: Object [_doc] has subobjects set to false hence it does not support inner object "
+                + "[metrics.service.time]",
+            exception.getMessage()
+        );
+    }
+
+    public void testSubobjectsCannotBeUpdated() throws IOException {
+        MapperService mapperService = createMapperService(fieldMapping(b -> b.field("type", "object")));
+        DocumentMapper mapper = mapperService.documentMapper();
+        assertNull(mapper.mapping().getRoot().dynamic());
+        Mapping mergeWith = mapperService.parseMapping("_doc", new CompressedXContent(BytesReference.bytes(fieldMapping(b -> {
+            b.field("type", "object");
+            b.field("subobjects", "false");
+        }))));
+        MapperException exception = expectThrows(
+            MapperException.class,
+            () -> mapper.mapping().merge(mergeWith, MergeReason.MAPPING_UPDATE)
+        );
+        assertEquals("the [subobjects] parameter can't be updated for the object mapping [field]", exception.getMessage());
+    }
+
+    public void testSubobjectsCannotBeUpdatedOnRoot() throws IOException {
+        MapperService mapperService = createMapperService(topMapping(b -> b.field("subobjects", false)));
+        DocumentMapper mapper = mapperService.documentMapper();
+        assertNull(mapper.mapping().getRoot().dynamic());
+        Mapping mergeWith = mapperService.parseMapping(
+            "_doc",
+            new CompressedXContent(BytesReference.bytes(topMapping(b -> { b.field("subobjects", true); })))
+        );
+        MapperException exception = expectThrows(
+            MapperException.class,
+            () -> mapper.mapping().merge(mergeWith, MergeReason.MAPPING_UPDATE)
+        );
+        assertEquals("the [subobjects] parameter can't be updated for the object mapping [_doc]", exception.getMessage());
+    }
 }

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

@@ -8,6 +8,8 @@
 
 package org.elasticsearch.index.mapper;
 
+import org.elasticsearch.xcontent.XContentBuilder;
+
 import java.io.IOException;
 
 import static org.hamcrest.Matchers.equalTo;
@@ -38,12 +40,42 @@ public class SourceLoaderTests extends MapperServiceTestCase {
 
     public void testDotsInFieldName() throws IOException {
         DocumentMapper mapper = createDocumentMapper(
-            syntheticSourceMapping(b -> { b.startObject("foo.bar.baz").field("type", "keyword").endObject(); })
+            syntheticSourceMapping(b -> b.startObject("foo.bar.baz").field("type", "keyword").endObject())
         );
         assertThat(syntheticSource(mapper, b -> b.field("foo.bar.baz", "aaa")), equalTo("""
             {"foo":{"bar":{"baz":"aaa"}}}"""));
     }
 
+    public void testNoSubobjectsIntermediateObject() throws IOException {
+        DocumentMapper mapper = createDocumentMapper(syntheticSourceMapping(b -> {
+            b.startObject("foo");
+            {
+                b.field("type", "object").field("subobjects", false);
+                b.startObject("properties");
+                {
+                    b.startObject("bar.baz").field("type", "keyword").endObject();
+                }
+                b.endObject();
+            }
+            b.endObject();
+        }));
+        assertThat(syntheticSource(mapper, b -> b.field("foo.bar.baz", "aaa")), equalTo("""
+            {"foo":{"bar.baz":"aaa"}}"""));
+    }
+
+    public void testNoSubobjectsRootObject() throws IOException {
+        XContentBuilder mappings = topMapping(b -> {
+            b.startObject("_source").field("synthetic", true).endObject();
+            b.field("subobjects", false);
+            b.startObject("properties");
+            b.startObject("foo.bar.baz").field("type", "keyword").endObject();
+            b.endObject();
+        });
+        DocumentMapper mapper = createDocumentMapper(mappings);
+        assertThat(syntheticSource(mapper, b -> b.field("foo.bar.baz", "aaa")), equalTo("""
+            {"foo.bar.baz":"aaa"}"""));
+    }
+
     public void testSorted() throws IOException {
         DocumentMapper mapper = createDocumentMapper(syntheticSourceMapping(b -> {
             b.startObject("foo").field("type", "keyword").endObject();