[8.x] Add object param for keeping synthetic source (#113690) (#114058)

* Add object param for keeping synthetic source (#113690)

* Add object param for keeping synthetic source

* Update docs/changelog/113690.yaml

* fix merging

* add tests

* merge

* fix randomized tests

* add documentation

* dedup id in docs

* update documentation

* update documentation

* fix bwc

* fix bwc

* fix unintended

* Revert "fix bwc"

This reverts commit 18dc913eee209b27c14f5cdaac4247c18b65c6b1.

* Revert "fix bwc"

This reverts commit f4ddb0e5e5bf26b841a3216aa5969ed60dabde59.

* add missing test

* fix transform

* fix transform

* fix transform

* fix transform

* fix transform

(cherry picked from commit dd2024881d038d64532b8eea35cac88142b503cb)

# Conflicts:
#	rest-api-spec/build.gradle

* Update build.gradle

* Update MapperFeatures.java

* Update 20_synthetic_source.yml

* Update 21_synthetic_source_stored.yml

* Update 21_synthetic_source_stored.yml

* Update 21_synthetic_source_stored.yml

* Update 21_synthetic_source_stored.yml

docs/changelog/113690.yaml

@@ -0,0 +1,5 @@
+pr: 113690
+summary: Add object param for keeping synthetic source
+area: Mapping
+type: enhancement
+issues: []

docs/reference/mapping/fields/synthetic-source.asciidoc

@@ -32,18 +32,25 @@ space. Additional latency can be avoided by not loading `_source` field in queri
 
 [[synthetic-source-fields]]
 ===== Supported fields
-Synthetic `_source` is supported by all field types. Depending on implementation details, field types have different properties when used with synthetic `_source`.
+Synthetic `_source` is supported by all field types. Depending on implementation details, field types have different
+properties when used with synthetic `_source`.
 
-<<synthetic-source-fields-native-list, Most field types>> construct synthetic `_source` using existing data, most commonly <<doc-values,`doc_values`>> and <<stored-fields, stored fields>>. For these field types, no additional space is needed to store the contents of `_source` field. Due to the storage layout of <<doc-values,`doc_values`>>, the generated `_source` field undergoes <<synthetic-source-modifications, modifications>> compared to original document.
+<<synthetic-source-fields-native-list, Most field types>> construct synthetic `_source` using existing data, most
+commonly <<doc-values,`doc_values`>> and <<stored-fields, stored fields>>. For these field types, no additional space
+is needed to store the contents of `_source` field. Due to the storage layout of <<doc-values,`doc_values`>>, the
+generated `_source` field undergoes <<synthetic-source-modifications, modifications>> compared to original document.
 
-For all other field types, the original value of the field is stored as is, in the same way as the `_source` field in non-synthetic mode. In this case there are no modifications and field data in `_source` is the same as in the original document. Similarly, malformed values of fields that use <<ignore-malformed,`ignore_malformed`>> or <<ignore-above,`ignore_above`>> need to be stored as is. This approach is less storage efficient since data needed for `_source` reconstruction is stored in addition to other data required to index the field (like `doc_values`).
+For all other field types, the original value of the field is stored as is, in the same way as the `_source` field in
+non-synthetic mode. In this case there are no modifications and field data in `_source` is the same as in the original
+document. Similarly, malformed values of fields that use <<ignore-malformed,`ignore_malformed`>> or
+<<ignore-above,`ignore_above`>> need to be stored as is. This approach is less storage efficient since data needed for
+`_source` reconstruction is stored in addition to other data required to index the field (like `doc_values`).
 
 [[synthetic-source-restrictions]]
 ===== Synthetic `_source` restrictions
 
-Synthetic `_source` cannot be used together with field mappings that use <<copy-to,`copy_to`>>. 
-
-Some field types have additional restrictions. These restrictions are documented in the **synthetic `_source`** section of the field type's <<mapping-types,documentation>>.
+Some field types have additional restrictions. These restrictions are documented in the **synthetic `_source`** section
+of the field type's <<mapping-types,documentation>>.
 
 [[synthetic-source-modifications]]
 ===== Synthetic `_source` modifications
@@ -144,6 +151,42 @@ Will become:
 ----
 // TEST[s/^/{"_source":/ s/\n$/}/]
 
+This impacts how source contents can be referenced in <<modules-scripting-using,scripts>>. For instance, referencing
+a script in its original source form will return null:
+
+[source,js]
+----
+"script": { "source": """  emit(params._source['foo.bar.baz'])  """ }
+----
+// NOTCONSOLE
+
+Instead, source references need to be in line with the mapping structure:
+
+[source,js]
+----
+"script": { "source": """  emit(params._source['foo']['bar']['baz'])  """ }
+----
+// NOTCONSOLE
+
+or simply
+
+[source,js]
+----
+"script": { "source": """  emit(params._source.foo.bar.baz)  """ }
+----
+// NOTCONSOLE
+
+The following <<modules-scripting-fields, field APIs>> are preferable as, in addition to being agnostic to the
+mapping structure, they make use of docvalues if available and fall back to synthetic source only when needed. This
+reduces source synthesizing, a slow and costly operation.
+
+[source,js]
+----
+"script": { "source": """  emit(field('foo.bar.baz').get(null))   """ }
+"script": { "source": """  emit($('foo.bar.baz', null))   """ }
+----
+// NOTCONSOLE
+
 [[synthetic-source-modifications-alphabetical]]
 ====== Alphabetical sorting
 Synthetic `_source` fields are sorted alphabetically. The
@@ -155,18 +198,99 @@ that ordering.
 
 [[synthetic-source-modifications-ranges]]
 ====== Representation of ranges
-Range field values (e.g. `long_range`) are always represented as inclusive on both sides with bounds adjusted accordingly. See <<range-synthetic-source-inclusive, examples>>.
+Range field values (e.g. `long_range`) are always represented as inclusive on both sides with bounds adjusted
+accordingly. See <<range-synthetic-source-inclusive, examples>>.
 
 [[synthetic-source-precision-loss-for-point-types]]
 ====== Reduced precision of `geo_point` values
-Values of `geo_point` fields are represented in synthetic `_source` with reduced precision. See <<geo-point-synthetic-source, examples>>.
+Values of `geo_point` fields are represented in synthetic `_source` with reduced precision. See
+<<geo-point-synthetic-source, examples>>.
+
+[[synthetic-source-keep]]
+====== Minimizing source modifications
+
+It is possible to avoid synthetic source modifications for a particular object or field, at extra storage cost.
+This is controlled through param `synthetic_source_keep` with the following option:
+
+ - `none`: synthetic source diverges from the original source as described above (default).
+ - `arrays`: arrays of the corresponding field or object preserve the original element ordering and duplicate elements.
+The synthetic source fragment for such arrays is not guaranteed to match the original source exactly, e.g. array
+`[1, 2, [5], [[4, [3]]], 5]` may appear as-is or in an equivalent format like `[1, 2, 5, 4, 3, 5]`. The exact format
+may change in the future, in an effort to reduce the storage overhead of this option.
+- `all`: the source for both singleton instances and arrays of the corresponding field or object gets recorded. When
+applied to objects, the source of all sub-objects and sub-fields gets captured. Furthermore, the original source of
+arrays gets captured and appears in synthetic source with no modifications.
+
+For instance:
+
+[source,console,id=create-index-with-synthetic-source-keep]
+----
+PUT idx_keep
+{
+  "mappings": {
+    "_source": {
+      "mode": "synthetic"
+    },
+    "properties": {
+      "path": {
+        "type": "object",
+        "synthetic_source_keep": "all"
+      },
+      "ids": {
+        "type": "integer",
+        "synthetic_source_keep": "arrays"
+      }
+    }
+  }
+}
+----
+// TEST
+
+[source,console,id=synthetic-source-keep-example]
+----
+PUT idx_keep/_doc/1
+{
+  "path": {
+    "to": [
+      { "foo": [3, 2, 1] },
+      { "foo": [30, 20, 10] }
+    ],
+    "bar": "baz"
+  },
+  "ids": [ 200, 100, 300, 100 ]
+}
+----
+// TEST[s/$/\nGET idx_keep\/_doc\/1?filter_path=_source\n/]
+
+returns the original source, with no array deduplication and sorting:
+
+[source,console-result]
+----
+{
+  "path": {
+    "to": [
+      { "foo": [3, 2, 1] },
+      { "foo": [30, 20, 10] }
+    ],
+    "bar": "baz"
+  },
+  "ids": [ 200, 100, 300, 100 ]
+}
+----
+// TEST[s/^/{"_source":/ s/\n$/}/]
 
+The option for capturing the source of arrays can be applied at index level, by setting
+`index.mapping.synthetic_source_keep` to `arrays`. This applies to all objects and fields in the index, except for
+the ones with explicit overrides of `synthetic_source_keep` set to `none`. In this case, the storage overhead grows
+with the number and sizes of arrays present in source of each document, naturally.
 
 [[synthetic-source-fields-native-list]]
 ===== Field types that support synthetic source with no storage overhead
-The following field types support synthetic source using data from <<doc-values,`doc_values`>> or <<stored-fields, stored fields>>, and require no additional storage space to construct the `_source` field. 
+The following field types support synthetic source using data from <<doc-values,`doc_values`>> or
+<stored-fields, stored fields>>, and require no additional storage space to construct the `_source` field.
 
-NOTE: If you enable the <<ignore-malformed,`ignore_malformed`>> or <<ignore-above,`ignore_above`>> settings, then additional storage is required to store ignored field values for these types.
+NOTE: If you enable the <<ignore-malformed,`ignore_malformed`>> or <<ignore-above,`ignore_above`>> settings, then
+additional storage is required to store ignored field values for these types.
 
 ** <<aggregate-metric-double-synthetic-source, `aggregate_metric_double`>>
 ** {plugins}/mapper-annotated-text-usage.html#annotated-text-synthetic-source[`annotated-text`]

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

@@ -920,7 +920,7 @@ subobjects auto:
                       id:
                         type: keyword
               stored:
-                store_array_source: true
+                synthetic_source_keep: arrays
                 properties:
                   span:
                     properties:

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/indices.create/21_synthetic_source_stored.yml

@@ -1,7 +1,71 @@
+---
+object param - store complex object:
+  - requires:
+      cluster_features: ["mapper.synthetic_source_keep"]
+      reason: requires tracking ignored source
+
+  - do:
+      indices.create:
+        index: test
+        body:
+          mappings:
+            _source:
+              mode: synthetic
+            properties:
+              id:
+                type: integer
+              stored:
+                synthetic_source_keep: all
+                properties:
+                  object_array:
+                    properties:
+                      trace:
+                        type: keyword
+                  nested:
+                    type: nested
+                  kw:
+                    type: keyword
+
+  - do:
+      bulk:
+        index: test
+        refresh: true
+        body:
+          - '{ "create": { } }'
+          - '{ "id": 1, "stored": { "object_array": [ {"trace": "B"}, {"trace": "A"} ], "nested": [ {"foo": 20}, {"foo": 10} ], "kw": 100 } }'
+          - '{ "create": { } }'
+          - '{ "id": 2, "stored": { "object_array": { "trace": ["D", "C"] }, "nested": { "bar": [ 40, 30] }, "kw": 200, "baz": "2000" } }'
+          - '{ "create": { } }'
+          - '{ "id": 3, "stored": [ { "object_array": { "trace": "E" } }, { "nested": { "bar": [ 60, 50] } }, { "kw": 300 } ] }'
+
+  - do:
+      search:
+        index: test
+        sort: id
+
+  - match: { hits.hits.0._source.id: 1 }
+  - match: { hits.hits.0._source.stored.object_array.0.trace: B }
+  - match: { hits.hits.0._source.stored.object_array.1.trace: A }
+  - match: { hits.hits.0._source.stored.nested.0.foo: 20 }
+  - match: { hits.hits.0._source.stored.nested.1.foo: 10 }
+  - match: { hits.hits.0._source.stored.kw: 100 }
+
+  - match: { hits.hits.1._source.id: 2 }
+  - match: { hits.hits.1._source.stored.object_array.trace: [D, C] }
+  - match: { hits.hits.1._source.stored.nested.bar: [40, 30] }
+  - match: { hits.hits.1._source.stored.kw: 200 }
+  - match: { hits.hits.1._source.stored.baz: "2000" }
+
+  - match: { hits.hits.2._source.id: 3 }
+  - match: { hits.hits.2._source.stored.0.object_array.trace: E }
+  - match: { hits.hits.2._source.stored.1.nested.bar: [ 60, 50 ] }
+  - match: { hits.hits.2._source.stored.2.kw: 300 }
+
+
 ---
 object param - object array:
   - requires:
-      cluster_features: ["mapper.track_ignored_source"]
+      cluster_features: ["mapper.synthetic_source_keep"]
       reason: requires tracking ignored source
 
   - do:
@@ -25,7 +89,7 @@ object param - object array:
                       id:
                         type: keyword
               stored:
-                store_array_source: true
+                synthetic_source_keep: arrays
                 properties:
                   span:
                     properties:
@@ -65,7 +129,7 @@ object param - object array:
 ---
 object param - object array within array:
   - requires:
-      cluster_features: ["mapper.track_ignored_source"]
+      cluster_features: ["mapper.synthetic_source_keep"]
       reason: requires tracking ignored source
 
   - do:
@@ -77,10 +141,10 @@ object param - object array within array:
               mode: synthetic
             properties:
               stored:
-                store_array_source: true
+                synthetic_source_keep: arrays
                 properties:
                   path:
-                    store_array_source: true
+                    synthetic_source_keep: arrays
                     properties:
                       to:
                         properties:
@@ -108,7 +172,7 @@ object param - object array within array:
 ---
 object param - no object array:
   - requires:
-      cluster_features: ["mapper.track_ignored_source"]
+      cluster_features: ["mapper.synthetic_source_keep"]
       reason: requires tracking ignored source
 
   - do:
@@ -120,7 +184,7 @@ object param - no object array:
               mode: synthetic
             properties:
               stored:
-                store_array_source: true
+                synthetic_source_keep: arrays
                 properties:
                   span:
                     properties:
@@ -150,7 +214,7 @@ object param - no object array:
 ---
 object param - field ordering in object array:
   - requires:
-      cluster_features: ["mapper.track_ignored_source"]
+      cluster_features: ["mapper.synthetic_source_keep"]
       reason: requires tracking ignored source
 
   - do:
@@ -164,7 +228,7 @@ object param - field ordering in object array:
               a:
                 type: keyword
               b:
-                store_array_source: true
+                synthetic_source_keep: arrays
                 properties:
                   aa:
                     type: keyword
@@ -173,7 +237,7 @@ object param - field ordering in object array:
               c:
                 type: keyword
               d:
-                store_array_source: true
+                synthetic_source_keep: arrays
                 properties:
                   aa:
                     type: keyword
@@ -199,7 +263,7 @@ object param - field ordering in object array:
 ---
 object param - nested object array next to other fields:
   - requires:
-      cluster_features: ["mapper.track_ignored_source"]
+      cluster_features: ["mapper.synthetic_source_keep"]
       reason: requires tracking ignored source
 
   - do:
@@ -215,7 +279,7 @@ object param - nested object array next to other fields:
               b:
                 properties:
                   c:
-                    store_array_source: true
+                    synthetic_source_keep: arrays
                     properties:
                       aa:
                         type: keyword
@@ -255,7 +319,7 @@ object param - nested object array next to other fields:
 ---
 object param - nested object with stored array:
   - requires:
-      cluster_features: ["mapper.track_ignored_source"]
+      cluster_features: ["mapper.synthetic_source_keep"]
       reason: requires tracking ignored source
 
   - do:
@@ -272,7 +336,7 @@ object param - nested object with stored array:
                 type: nested
               nested_array_stored:
                 type: nested
-                store_array_source: true
+                synthetic_source_keep: all
 
   - do:
       bulk:
@@ -322,7 +386,7 @@ index param - nested array within array:
                   to:
                     properties:
                       some:
-                        store_array_source: true
+                        synthetic_source_keep: arrays
                         properties:
                           id:
                             type: integer
@@ -351,7 +415,7 @@ index param - nested array within array:
 # 112156
 stored field under object with store_array_source:
   - requires:
-      cluster_features: ["mapper.source.synthetic_source_stored_fields_advance_fix"]
+      cluster_features: ["mapper.synthetic_source_keep"]
       reason: requires bug fix to be implemented
 
   - do:
@@ -369,7 +433,7 @@ stored field under object with store_array_source:
               name:
                 type: keyword
               obj:
-                store_array_source: true
+                synthetic_source_keep: arrays
                 properties:
                   foo:
                     type: keyword
@@ -772,6 +836,9 @@ index param - root arrays:
                     properties:
                       id:
                         type: keyword
+              obj_default:
+                type: object
+                synthetic_source_keep: none
 
   - do:
       bulk:
@@ -782,6 +849,8 @@ index param - root arrays:
           - '{  "id": 1, "leaf": [30, 20, 10], "leaf_default": [30, 20, 10], "obj": [ { "trace": { "id": "a" }, "span": { "id": "1" } }, { "trace": { "id": "b" }, "span": { "id": "1" } } ] }'
           - '{ "create": { } }'
           - '{  "id": 2, "leaf": [130, 120, 110], "leaf_default": [130, 120, 110], "obj": [ { "trace": { "id": "aa" }, "span": { "id": "2" } }, { "trace": { "id": "bb" }, "span": { "id": "2" } } ] }'
+          - '{ "create": { } }'
+          - '{  "id": 3, "obj_default": [ { "trace": { "id": "bb" }, "span": { "id": "2" } }, { "trace": { "id": "aa" }, "span": { "id": "2" } } ] }'
 
   - do:
       search:
@@ -799,13 +868,17 @@ index param - root arrays:
 
   - match: { hits.hits.1._source.id: 2 }
   - match: { hits.hits.1._source.leaf: [ 130, 120, 110 ] }
-  - match: { hits.hits.0._source.leaf_default: [10, 20, 30]  }
+  - match: { hits.hits.1._source.leaf_default: [110, 120, 130]  }
   - length: { hits.hits.1._source.obj: 2 }
   - match: { hits.hits.1._source.obj.0.trace.id: aa }
   - match: { hits.hits.1._source.obj.0.span.id: "2" }
   - match: { hits.hits.1._source.obj.1.trace.id: bb }
   - match: { hits.hits.1._source.obj.1.span.id: "2" }
 
+  - match: { hits.hits.2._source.id: 3 }
+  - match: { hits.hits.2._source.obj_default.trace.id: [aa, bb] }
+  - match: { hits.hits.2._source.obj_default.span.id: "2" }
+
 
 ---
 index param - dynamic root arrays:

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

@@ -421,20 +421,21 @@ public final class DocumentParser {
             throwOnConcreteValue(context.parent(), currentFieldName, context);
         }
 
+        if (context.canAddIgnoredField() && getSourceKeepMode(context, context.parent().sourceKeepMode()) == Mapper.SourceKeepMode.ALL) {
+            context = context.addIgnoredFieldFromContext(
+                new IgnoredSourceFieldMapper.NameValue(
+                    context.parent().fullPath(),
+                    context.parent().fullPath().lastIndexOf(context.parent().leafName()),
+                    null,
+                    context.doc()
+                )
+            );
+            token = context.parser().currentToken();
+            parser = context.parser();
+        }
+
         if (context.parent().isNested()) {
             // Handle a nested object that doesn't contain an array. Arrays are handled in #parseNonDynamicArray.
-            if (context.parent().storeArraySource() && context.canAddIgnoredField()) {
-                context = context.addIgnoredFieldFromContext(
-                    new IgnoredSourceFieldMapper.NameValue(
-                        context.parent().fullPath(),
-                        context.parent().fullPath().lastIndexOf(context.parent().leafName()),
-                        null,
-                        context.doc()
-                    )
-                );
-                token = context.parser().currentToken();
-                parser = context.parser();
-            }
             context = context.createNestedContext((NestedObjectMapper) context.parent());
         }
 
@@ -801,8 +802,8 @@ public final class DocumentParser {
         // Check if we need to record the array source. This only applies to synthetic source.
         if (context.canAddIgnoredField()) {
             boolean objectRequiresStoringSource = mapper instanceof ObjectMapper objectMapper
-                && (objectMapper.storeArraySource()
-                    || (context.sourceKeepModeFromIndexSettings() == Mapper.SourceKeepMode.ARRAYS
+                && (getSourceKeepMode(context, objectMapper.sourceKeepMode()) == Mapper.SourceKeepMode.ALL
+                    || (getSourceKeepMode(context, objectMapper.sourceKeepMode()) == Mapper.SourceKeepMode.ARRAYS
                         && objectMapper instanceof NestedObjectMapper == false));
             boolean fieldWithFallbackSyntheticSource = mapper instanceof FieldMapper fieldMapper
                 && fieldMapper.syntheticSourceMode() == FieldMapper.SyntheticSourceMode.FALLBACK;
@@ -1115,15 +1116,7 @@ public final class DocumentParser {
 
     private static class NoOpObjectMapper extends ObjectMapper {
         NoOpObjectMapper(String name, String fullPath) {
-            super(
-                name,
-                fullPath,
-                Explicit.IMPLICIT_TRUE,
-                Optional.empty(),
-                Explicit.IMPLICIT_FALSE,
-                Dynamic.RUNTIME,
-                Collections.emptyMap()
-            );
+            super(name, fullPath, Explicit.IMPLICIT_TRUE, Optional.empty(), Optional.empty(), Dynamic.RUNTIME, Collections.emptyMap());
         }
 
         @Override

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

@@ -98,6 +98,17 @@ public class NestedObjectMapper extends ObjectMapper {
             } else {
                 nestedTypePath = fullPath;
             }
+            if (sourceKeepMode.orElse(SourceKeepMode.NONE) == SourceKeepMode.ARRAYS) {
+                throw new MapperException(
+                    "parameter [ "
+                        + Mapper.SYNTHETIC_SOURCE_KEEP_PARAM
+                        + " ] can't be set to ["
+                        + SourceKeepMode.ARRAYS
+                        + "] for nested object ["
+                        + fullPath
+                        + "]"
+                );
+            }
             final Query nestedTypeFilter = NestedPathFieldMapper.filter(indexCreatedVersion, nestedTypePath);
             NestedMapperBuilderContext nestedContext = new NestedMapperBuilderContext(
                 context.buildFullName(leafName()),
@@ -115,7 +126,7 @@ public class NestedObjectMapper extends ObjectMapper {
                 buildMappers(nestedContext),
                 enabled,
                 dynamic,
-                storeArraySource,
+                sourceKeepMode,
                 includeInParent,
                 includeInRoot,
                 parentTypeFilter,
@@ -213,7 +224,7 @@ public class NestedObjectMapper extends ObjectMapper {
         Map<String, Mapper> mappers,
         Explicit<Boolean> enabled,
         ObjectMapper.Dynamic dynamic,
-        Explicit<Boolean> storeArraySource,
+        Optional<SourceKeepMode> sourceKeepMode,
         Explicit<Boolean> includeInParent,
         Explicit<Boolean> includeInRoot,
         Query parentTypeFilter,
@@ -222,7 +233,7 @@ public class NestedObjectMapper extends ObjectMapper {
         Function<Query, BitSetProducer> bitsetProducer,
         IndexSettings indexSettings
     ) {
-        super(name, fullPath, enabled, Optional.empty(), storeArraySource, dynamic, mappers);
+        super(name, fullPath, enabled, Optional.empty(), sourceKeepMode, dynamic, mappers);
         this.parentTypeFilter = parentTypeFilter;
         this.nestedTypePath = nestedTypePath;
         this.nestedTypeFilter = nestedTypeFilter;
@@ -283,7 +294,7 @@ public class NestedObjectMapper extends ObjectMapper {
             Map.of(),
             enabled,
             dynamic,
-            storeArraySource,
+            sourceKeepMode,
             includeInParent,
             includeInRoot,
             parentTypeFilter,
@@ -310,8 +321,8 @@ public class NestedObjectMapper extends ObjectMapper {
         if (isEnabled() != Defaults.ENABLED) {
             builder.field("enabled", enabled.value());
         }
-        if (storeArraySource != Defaults.STORE_ARRAY_SOURCE) {
-            builder.field(STORE_ARRAY_SOURCE_PARAM, storeArraySource.value());
+        if (sourceKeepMode.isPresent()) {
+            builder.field(Mapper.SYNTHETIC_SOURCE_KEEP_PARAM, sourceKeepMode.get());
         }
         serializeMappers(builder, params);
         return builder.endObject();
@@ -359,7 +370,7 @@ public class NestedObjectMapper extends ObjectMapper {
             mergeResult.mappers(),
             mergeResult.enabled(),
             mergeResult.dynamic(),
-            mergeResult.trackArraySource(),
+            mergeResult.sourceKeepMode(),
             incInParent,
             incInRoot,
             parentTypeFilter,
@@ -393,8 +404,8 @@ public class NestedObjectMapper extends ObjectMapper {
 
     @Override
     public SourceLoader.SyntheticFieldLoader syntheticFieldLoader() {
-        if (storeArraySource()) {
-            // IgnoredSourceFieldMapper integration takes care of writing the source for nested objects that enabled store_array_source.
+        if (sourceKeepMode.orElse(SourceKeepMode.NONE) == SourceKeepMode.ALL) {
+            // IgnoredSourceFieldMapper integration takes care of writing the source for the nested object.
             return SourceLoader.SyntheticFieldLoader.NOTHING;
         }
 

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");
+    // No-op. All uses of this feature were reverted but node features can't be removed.
     static final NodeFeature SUBOBJECTS_AUTO_FIXES = new NodeFeature("mapper.subobjects_auto_fixes");
 
     /**
@@ -127,7 +128,7 @@ public class ObjectMapper extends Mapper {
     public static class Builder extends Mapper.Builder {
         protected Optional<Subobjects> subobjects;
         protected Explicit<Boolean> enabled = Explicit.IMPLICIT_TRUE;
-        protected Explicit<Boolean> storeArraySource = Defaults.STORE_ARRAY_SOURCE;
+        protected Optional<SourceKeepMode> sourceKeepMode = Optional.empty();
         protected Dynamic dynamic;
         protected final List<Mapper.Builder> mappersBuilders = new ArrayList<>();
 
@@ -141,8 +142,8 @@ public class ObjectMapper extends Mapper {
             return this;
         }
 
-        public Builder storeArraySource(boolean value) {
-            this.storeArraySource = Explicit.explicitBoolean(value);
+        public Builder sourceKeepMode(SourceKeepMode sourceKeepMode) {
+            this.sourceKeepMode = Optional.of(sourceKeepMode);
             return this;
         }
 
@@ -245,7 +246,7 @@ public class ObjectMapper extends Mapper {
                 context.buildFullName(leafName()),
                 enabled,
                 subobjects,
-                storeArraySource,
+                sourceKeepMode,
                 dynamic,
                 buildMappers(context.createChildContext(leafName(), dynamic))
             );
@@ -307,7 +308,10 @@ public class ObjectMapper extends Mapper {
                 builder.enabled(XContentMapValues.nodeBooleanValue(fieldNode, fieldName + ".enabled"));
                 return true;
             } else if (fieldName.equals(STORE_ARRAY_SOURCE_PARAM)) {
-                builder.storeArraySource(XContentMapValues.nodeBooleanValue(fieldNode, fieldName + ".store_array_source"));
+                builder.sourceKeepMode(SourceKeepMode.ARRAYS);
+                return true;
+            } else if (fieldName.equals(Mapper.SYNTHETIC_SOURCE_KEEP_PARAM)) {
+                builder.sourceKeepMode(SourceKeepMode.from(fieldNode.toString()));
                 return true;
             } else if (fieldName.equals("properties")) {
                 if (fieldNode instanceof Collection && ((Collection) fieldNode).isEmpty()) {
@@ -434,7 +438,7 @@ public class ObjectMapper extends Mapper {
 
     protected final Explicit<Boolean> enabled;
     protected final Optional<Subobjects> subobjects;
-    protected final Explicit<Boolean> storeArraySource;
+    protected final Optional<SourceKeepMode> sourceKeepMode;
     protected final Dynamic dynamic;
 
     protected final Map<String, Mapper> mappers;
@@ -444,7 +448,7 @@ public class ObjectMapper extends Mapper {
         String fullPath,
         Explicit<Boolean> enabled,
         Optional<Subobjects> subobjects,
-        Explicit<Boolean> storeArraySource,
+        Optional<SourceKeepMode> sourceKeepMode,
         Dynamic dynamic,
         Map<String, Mapper> mappers
     ) {
@@ -454,7 +458,7 @@ public class ObjectMapper extends Mapper {
         this.fullPath = internFieldName(fullPath);
         this.enabled = enabled;
         this.subobjects = subobjects;
-        this.storeArraySource = storeArraySource;
+        this.sourceKeepMode = sourceKeepMode;
         this.dynamic = dynamic;
         if (mappers == null) {
             this.mappers = Map.of();
@@ -482,7 +486,7 @@ public class ObjectMapper extends Mapper {
      * This is typically used in the context of a mapper merge when there's not enough budget to add the entire object.
      */
     ObjectMapper withoutMappers() {
-        return new ObjectMapper(leafName(), fullPath, enabled, subobjects, storeArraySource, dynamic, Map.of());
+        return new ObjectMapper(leafName(), fullPath, enabled, subobjects, sourceKeepMode, dynamic, Map.of());
     }
 
     @Override
@@ -520,8 +524,8 @@ public class ObjectMapper extends Mapper {
         return subobjects.orElse(Subobjects.ENABLED);
     }
 
-    public final boolean storeArraySource() {
-        return storeArraySource.value();
+    public final Optional<SourceKeepMode> sourceKeepMode() {
+        return sourceKeepMode;
     }
 
     @Override
@@ -550,7 +554,7 @@ public class ObjectMapper extends Mapper {
             fullPath,
             mergeResult.enabled,
             mergeResult.subObjects,
-            mergeResult.trackArraySource,
+            mergeResult.sourceKeepMode,
             mergeResult.dynamic,
             mergeResult.mappers
         );
@@ -559,7 +563,7 @@ public class ObjectMapper extends Mapper {
     protected record MergeResult(
         Explicit<Boolean> enabled,
         Optional<Subobjects> subObjects,
-        Explicit<Boolean> trackArraySource,
+        Optional<SourceKeepMode> sourceKeepMode,
         Dynamic dynamic,
         Map<String, Mapper> mappers
     ) {
@@ -593,26 +597,31 @@ public class ObjectMapper extends Mapper {
             } else {
                 subObjects = existing.subobjects;
             }
-            final Explicit<Boolean> trackArraySource;
-            if (mergeWithObject.storeArraySource.explicit()) {
+            final Optional<SourceKeepMode> sourceKeepMode;
+            if (mergeWithObject.sourceKeepMode.isPresent()) {
                 if (reason == MergeReason.INDEX_TEMPLATE) {
-                    trackArraySource = mergeWithObject.storeArraySource;
-                } else if (existing.storeArraySource != mergeWithObject.storeArraySource) {
+                    sourceKeepMode = mergeWithObject.sourceKeepMode;
+                } else if (existing.sourceKeepMode.isEmpty() || existing.sourceKeepMode.get() != mergeWithObject.sourceKeepMode.get()) {
                     throw new MapperException(
-                        "the [store_array_source] parameter can't be updated for the object mapping [" + existing.fullPath() + "]"
+                        "the [ "
+                            + Mapper.SYNTHETIC_SOURCE_KEEP_PARAM
+                            + " ] parameter can't be updated for the object mapping ["
+                            + existing.fullPath()
+                            + "]"
                     );
                 } else {
-                    trackArraySource = existing.storeArraySource;
+                    sourceKeepMode = existing.sourceKeepMode;
                 }
             } else {
-                trackArraySource = existing.storeArraySource;
+                sourceKeepMode = existing.sourceKeepMode;
             }
+
             MapperMergeContext objectMergeContext = existing.createChildContext(parentMergeContext, existing.leafName());
             Map<String, Mapper> mergedMappers = buildMergedMappers(existing, mergeWithObject, objectMergeContext, subObjects);
             return new MergeResult(
                 enabled,
                 subObjects,
-                trackArraySource,
+                sourceKeepMode,
                 mergeWithObject.dynamic != null ? mergeWithObject.dynamic : existing.dynamic,
                 mergedMappers
             );
@@ -733,6 +742,12 @@ public class ObjectMapper extends Mapper {
                     + ")"
             );
         }
+        if (sourceKeepMode.isPresent()) {
+            throwAutoFlatteningException(
+                path,
+                "the value of [" + Mapper.SYNTHETIC_SOURCE_KEEP_PARAM + "] is [ " + sourceKeepMode.get() + " ]"
+            );
+        }
         if (isEnabled() == false) {
             throwAutoFlatteningException(path, "the value of [enabled] is [false]");
         }
@@ -774,8 +789,8 @@ public class ObjectMapper extends Mapper {
         if (subobjects.isPresent()) {
             builder.field("subobjects", subobjects.get().printedValue);
         }
-        if (storeArraySource != Defaults.STORE_ARRAY_SOURCE) {
-            builder.field(STORE_ARRAY_SOURCE_PARAM, storeArraySource.value());
+        if (sourceKeepMode.isPresent()) {
+            builder.field("synthetic_source_keep", sourceKeepMode.get());
         }
         if (custom != null) {
             custom.toXContent(builder, params);

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

@@ -82,6 +82,7 @@ public class PassThroughObjectMapper extends ObjectMapper {
                 leafName(),
                 context.buildFullName(leafName()),
                 enabled,
+                sourceKeepMode,
                 dynamic,
                 buildMappers(context.createChildContext(leafName(), timeSeriesDimensionSubFields.value(), dynamic)),
                 timeSeriesDimensionSubFields,
@@ -99,13 +100,14 @@ public class PassThroughObjectMapper extends ObjectMapper {
         String name,
         String fullPath,
         Explicit<Boolean> enabled,
+        Optional<SourceKeepMode> sourceKeepMode,
         Dynamic dynamic,
         Map<String, Mapper> mappers,
         Explicit<Boolean> timeSeriesDimensionSubFields,
         int priority
     ) {
         // Subobjects are not currently supported.
-        super(name, fullPath, enabled, Optional.of(Subobjects.DISABLED), Explicit.IMPLICIT_FALSE, dynamic, mappers);
+        super(name, fullPath, enabled, Optional.of(Subobjects.DISABLED), sourceKeepMode, dynamic, mappers);
         this.timeSeriesDimensionSubFields = timeSeriesDimensionSubFields;
         this.priority = priority;
         if (priority < 0) {
@@ -115,7 +117,16 @@ public class PassThroughObjectMapper extends ObjectMapper {
 
     @Override
     PassThroughObjectMapper withoutMappers() {
-        return new PassThroughObjectMapper(leafName(), fullPath(), enabled, dynamic, Map.of(), timeSeriesDimensionSubFields, priority);
+        return new PassThroughObjectMapper(
+            leafName(),
+            fullPath(),
+            enabled,
+            sourceKeepMode,
+            dynamic,
+            Map.of(),
+            timeSeriesDimensionSubFields,
+            priority
+        );
     }
 
     @Override
@@ -158,6 +169,7 @@ public class PassThroughObjectMapper extends ObjectMapper {
             leafName(),
             fullPath(),
             mergeResult.enabled(),
+            mergeResult.sourceKeepMode(),
             mergeResult.dynamic(),
             mergeResult.mappers(),
             containsDimensions,

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

@@ -113,7 +113,7 @@ public class RootObjectMapper extends ObjectMapper {
                 leafName(),
                 enabled,
                 subobjects,
-                storeArraySource,
+                sourceKeepMode,
                 dynamic,
                 buildMappers(context.createChildContext(null, dynamic)),
                 new HashMap<>(runtimeFields),
@@ -135,7 +135,7 @@ public class RootObjectMapper extends ObjectMapper {
         String name,
         Explicit<Boolean> enabled,
         Optional<Subobjects> subobjects,
-        Explicit<Boolean> trackArraySource,
+        Optional<SourceKeepMode> sourceKeepMode,
         Dynamic dynamic,
         Map<String, Mapper> mappers,
         Map<String, RuntimeField> runtimeFields,
@@ -144,12 +144,17 @@ public class RootObjectMapper extends ObjectMapper {
         Explicit<Boolean> dateDetection,
         Explicit<Boolean> numericDetection
     ) {
-        super(name, name, enabled, subobjects, trackArraySource, dynamic, mappers);
+        super(name, name, enabled, subobjects, sourceKeepMode, dynamic, mappers);
         this.runtimeFields = runtimeFields;
         this.dynamicTemplates = dynamicTemplates;
         this.dynamicDateTimeFormatters = dynamicDateTimeFormatters;
         this.dateDetection = dateDetection;
         this.numericDetection = numericDetection;
+        if (sourceKeepMode.orElse(SourceKeepMode.NONE) == SourceKeepMode.ALL) {
+            throw new MapperParsingException(
+                "root object can't be configured with [" + Mapper.SYNTHETIC_SOURCE_KEEP_PARAM + ":" + SourceKeepMode.ALL + "]"
+            );
+        }
     }
 
     @Override
@@ -166,7 +171,7 @@ public class RootObjectMapper extends ObjectMapper {
             leafName(),
             enabled,
             subobjects,
-            storeArraySource,
+            sourceKeepMode,
             dynamic,
             Map.of(),
             Map.of(),
@@ -282,7 +287,7 @@ public class RootObjectMapper extends ObjectMapper {
             leafName(),
             mergeResult.enabled(),
             mergeResult.subObjects(),
-            mergeResult.trackArraySource(),
+            mergeResult.sourceKeepMode(),
             mergeResult.dynamic(),
             mergeResult.mappers(),
             Map.copyOf(runtimeFields),

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

@@ -179,7 +179,7 @@ public class FieldAliasMapperValidationTests extends ESTestCase {
             name,
             Explicit.IMPLICIT_TRUE,
             Optional.empty(),
-            Explicit.IMPLICIT_FALSE,
+            Optional.empty(),
             ObjectMapper.Dynamic.FALSE,
             emptyMap()
         );

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

@@ -20,6 +20,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 import static java.util.Collections.emptyList;
@@ -433,6 +434,7 @@ public class FieldTypeLookupTests extends ESTestCase {
             name,
             name,
             Explicit.EXPLICIT_TRUE,
+            Optional.empty(),
             ObjectMapper.Dynamic.FALSE,
             mappers,
             Explicit.EXPLICIT_FALSE,

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

@@ -578,13 +578,39 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
         })).documentMapper();
         var syntheticSource = syntheticSource(documentMapper, b -> {
             b.startArray("path");
+            b.startObject().field("int_value", 20).endObject();
             b.startObject().field("int_value", 10).endObject();
+            b.endArray();
+            b.field("bool_value", true);
+        });
+        assertEquals("""
+            {"bool_value":true,"path":[{"int_value":20},{"int_value":10}]}""", syntheticSource);
+    }
+
+    public void testIndexStoredArraySourceRootObjectArrayWithBypass() throws IOException {
+        DocumentMapper documentMapper = createMapperServiceWithStoredArraySource(syntheticSourceMapping(b -> {
+            b.startObject("path");
+            {
+                b.field("type", "object");
+                b.field("synthetic_source_keep", "none");
+                b.startObject("properties");
+                {
+                    b.startObject("int_value").field("type", "integer").endObject();
+                }
+                b.endObject();
+            }
+            b.endObject();
+            b.startObject("bool_value").field("type", "boolean").endObject();
+        })).documentMapper();
+        var syntheticSource = syntheticSource(documentMapper, b -> {
+            b.startArray("path");
             b.startObject().field("int_value", 20).endObject();
+            b.startObject().field("int_value", 10).endObject();
             b.endArray();
             b.field("bool_value", true);
         });
         assertEquals("""
-            {"bool_value":true,"path":[{"int_value":10},{"int_value":20}]}""", syntheticSource);
+            {"bool_value":true,"path":{"int_value":[10,20]}}""", syntheticSource);
     }
 
     public void testIndexStoredArraySourceNestedValueArray() throws IOException {
@@ -622,6 +648,12 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
                 {
                     b.startObject("int_value").field("type", "integer").field(Mapper.SYNTHETIC_SOURCE_KEEP_PARAM, "none").endObject();
                     b.startObject("bool_value").field("type", "boolean").endObject();
+                    b.startObject("obj").field("type", "object").field(Mapper.SYNTHETIC_SOURCE_KEEP_PARAM, "none");
+                    b.startObject("properties");
+                    {
+                        b.startObject("foo").field("type", "integer").endObject();
+                    }
+                    b.endObject().endObject();
                 }
                 b.endObject();
             }
@@ -632,11 +664,17 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
             {
                 b.array("int_value", new int[] { 30, 20, 10 });
                 b.field("bool_value", true);
+                b.startArray("obj");
+                {
+                    b.startObject().field("foo", 2).endObject();
+                    b.startObject().field("foo", 1).endObject();
+                }
+                b.endArray();
             }
             b.endObject();
         });
         assertEquals("""
-            {"path":{"bool_value":true,"int_value":[10,20,30]}}""", syntheticSource);
+            {"path":{"bool_value":true,"int_value":[10,20,30],"obj":{"foo":[1,2]}}}""", syntheticSource);
     }
 
     public void testFieldStoredArraySourceNestedValueArray() throws IOException {
@@ -674,8 +712,8 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
                 b.field("type", "object");
                 b.startObject("properties");
                 {
-                    b.startObject("default").field("type", "float").field(Mapper.SYNTHETIC_SOURCE_KEEP_PARAM, "none").endObject();
-                    b.startObject("source_kept").field("type", "float").field(Mapper.SYNTHETIC_SOURCE_KEEP_PARAM, "all").endObject();
+                    b.startObject("default").field("type", "float").field("synthetic_source_keep", "none").endObject();
+                    b.startObject("source_kept").field("type", "float").field("synthetic_source_keep", "all").endObject();
                     b.startObject("bool_value").field("type", "boolean").endObject();
                 }
                 b.endObject();
@@ -738,7 +776,7 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
             b.startObject("path");
             {
                 b.field("type", "object");
-                b.field("store_array_source", true);
+                b.field("synthetic_source_keep", "arrays");
                 b.startObject("properties");
                 {
                     b.startObject("int_value").field("type", "integer").endObject();
@@ -765,7 +803,7 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
                 b.field("type", "object");
                 b.startObject("properties");
                 {
-                    b.startObject("to").field("type", "object").field("store_array_source", true);
+                    b.startObject("to").field("type", "object").field("synthetic_source_keep", "arrays");
                     {
                         b.startObject("properties");
                         {
@@ -835,10 +873,10 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
         DocumentMapper documentMapper = createMapperService(syntheticSourceMapping(b -> {
             b.startObject("path");
             {
-                b.field("type", "object").field("store_array_source", true);
+                b.field("type", "object").field("synthetic_source_keep", "arrays");
                 b.startObject("properties");
                 {
-                    b.startObject("to").field("type", "object").field("store_array_source", true);
+                    b.startObject("to").field("type", "object").field("synthetic_source_keep", "arrays");
                     {
                         b.startObject("properties");
                         {
@@ -893,7 +931,7 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
                 {
                     b.startObject("stored");
                     {
-                        b.field("type", "object").field("store_array_source", true);
+                        b.field("type", "object").field("synthetic_source_keep", "arrays");
                         b.startObject("properties").startObject("leaf").field("type", "integer").endObject().endObject();
                     }
                     b.endObject();
@@ -1061,7 +1099,7 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
                 b.field("type", "object");
                 b.startObject("properties");
                 {
-                    b.startObject("to").field("type", "object").field("store_array_source", true);
+                    b.startObject("to").field("type", "object").field("synthetic_source_keep", "arrays");
                     {
                         b.startObject("properties");
                         {
@@ -1107,6 +1145,42 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
             {"path":{"to":[{"name":"A"},{"name":"B"},{"name":"C"},{"name":"D"}]}}""", booleanValue), syntheticSource);
     }
 
+    public void testObjectWithKeepAll() throws IOException {
+        DocumentMapper documentMapper = createMapperService(syntheticSourceMapping(b -> {
+            b.startObject("path");
+            {
+                b.field("type", "object").field("synthetic_source_keep", "all");
+                b.startObject("properties");
+                {
+                    b.startObject("a").field("type", "object").endObject();
+                    b.startObject("b").field("type", "integer").endObject();
+                }
+                b.endObject();
+            }
+            b.endObject();
+            b.startObject("id").field("type", "integer").endObject();
+        })).documentMapper();
+        var syntheticSource = syntheticSource(documentMapper, b -> {
+            b.startObject("path");
+            {
+                b.startArray("a");
+                {
+                    b.startObject().field("foo", 30).endObject();
+                    b.startObject().field("foo", 20).endObject();
+                    b.startObject().field("foo", 10).endObject();
+                    b.startObject().field("bar", 20).endObject();
+                    b.startObject().field("bar", 10).endObject();
+                }
+                b.endArray();
+                b.array("b", 4, 1, 3, 2);
+            }
+            b.endObject();
+            b.field("id", 10);
+        });
+        assertEquals("""
+            {"id":10,"path":{"a":[{"foo":30},{"foo":20},{"foo":10},{"bar":20},{"bar":10}],"b":[4,1,3,2]}}""", syntheticSource);
+    }
+
     public void testFallbackFieldWithinHigherLevelArray() throws IOException {
         DocumentMapper documentMapper = createMapperService(syntheticSourceMapping(b -> {
             b.startObject("path");
@@ -1140,7 +1214,7 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
     public void testFieldOrdering() throws IOException {
         DocumentMapper documentMapper = createMapperService(syntheticSourceMapping(b -> {
             b.startObject("A").field("type", "integer").endObject();
-            b.startObject("B").field("type", "object").field("store_array_source", true);
+            b.startObject("B").field("type", "object").field("synthetic_source_keep", "arrays");
             {
                 b.startObject("properties");
                 {
@@ -1151,7 +1225,7 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
             }
             b.endObject();
             b.startObject("C").field("type", "integer").endObject();
-            b.startObject("D").field("type", "object").field("store_array_source", true);
+            b.startObject("D").field("type", "object").field("synthetic_source_keep", "arrays");
             {
                 b.startObject("properties");
                 {
@@ -1189,7 +1263,7 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
         DocumentMapper documentMapper = createMapperService(syntheticSourceMapping(b -> {
             b.startObject("path").field("type", "nested");
             {
-                b.field("store_array_source", true);
+                b.field("synthetic_source_keep", "all");
                 b.startObject("properties");
                 {
                     b.startObject("foo").field("type", "keyword").endObject();
@@ -1211,7 +1285,7 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
         DocumentMapper documentMapper = createMapperService(syntheticSourceMapping(b -> {
             b.startObject("path").field("type", "nested");
             {
-                b.field("store_array_source", true);
+                b.field("synthetic_source_keep", "all");
                 b.startObject("properties");
                 {
                     b.startObject("foo").field("type", "keyword").endObject();
@@ -1244,7 +1318,7 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
                     b.startObject("int_value").field("type", "integer").endObject();
                     b.startObject("to").field("type", "nested");
                     {
-                        b.field("store_array_source", true);
+                        b.field("synthetic_source_keep", "all");
                         b.startObject("properties");
                         {
                             b.startObject("foo").field("type", "keyword").endObject();
@@ -1285,7 +1359,7 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
                     b.startObject("int_value").field("type", "integer").endObject();
                     b.startObject("to").field("type", "nested");
                     {
-                        b.field("store_array_source", true);
+                        b.field("synthetic_source_keep", "all");
                         b.startObject("properties");
                         {
                             b.startObject("foo").field("type", "keyword").endObject();
@@ -1325,7 +1399,7 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
 
     public void testNestedObjectIncludeInRoot() throws IOException {
         DocumentMapper documentMapper = createMapperService(syntheticSourceMapping(b -> {
-            b.startObject("path").field("type", "nested").field("store_array_source", true).field("include_in_root", true);
+            b.startObject("path").field("type", "nested").field("synthetic_source_keep", "all").field("include_in_root", true);
             {
                 b.startObject("properties");
                 {
@@ -1599,7 +1673,7 @@ public class IgnoredSourceFieldMapperTests extends MapperServiceTestCase {
             b.startObject("path");
             b.startObject("properties");
             {
-                b.startObject("at").field("type", "nested").field("store_array_source", "true").endObject();
+                b.startObject("at").field("type", "nested").field("synthetic_source_keep", "all").endObject();
             }
             b.endObject();
             b.endObject();

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

@@ -84,7 +84,7 @@ public class MappingLookupTests extends ESTestCase {
             "object",
             Explicit.EXPLICIT_TRUE,
             Optional.empty(),
-            Explicit.IMPLICIT_FALSE,
+            Optional.empty(),
             ObjectMapper.Dynamic.TRUE,
             Collections.singletonMap("object.subfield", fieldMapper)
         );

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

@@ -1571,14 +1571,14 @@ public class NestedObjectMapperTests extends MapperServiceTestCase {
 
     public void testStoreArraySourceinSyntheticSourceMode() throws IOException {
         DocumentMapper mapper = createDocumentMapper(syntheticSourceMapping(b -> {
-            b.startObject("o").field("type", "nested").field(ObjectMapper.STORE_ARRAY_SOURCE_PARAM, true).endObject();
+            b.startObject("o").field("type", "nested").field("synthetic_source_keep", "all").endObject();
         }));
         assertNotNull(mapper.mapping().getRoot().getMapper("o"));
     }
 
     public void testStoreArraySourceNoopInNonSyntheticSourceMode() throws IOException {
         DocumentMapper mapper = createDocumentMapper(mapping(b -> {
-            b.startObject("o").field("type", "nested").field(ObjectMapper.STORE_ARRAY_SOURCE_PARAM, true).endObject();
+            b.startObject("o").field("type", "nested").field("synthetic_source_keep", "all").endObject();
         }));
         assertNotNull(mapper.mapping().getRoot().getMapper("o"));
     }

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

@@ -167,7 +167,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
         assertNotNull(objectMapper);
         assertFalse(objectMapper.isEnabled());
         assertEquals(ObjectMapper.Subobjects.ENABLED, objectMapper.subobjects());
-        assertFalse(objectMapper.storeArraySource());
+        assertTrue(objectMapper.sourceKeepMode().isEmpty());
 
         // Setting 'enabled' to true is allowed, and updates the mapping.
         update = Strings.toString(
@@ -189,7 +189,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
         assertNotNull(objectMapper);
         assertTrue(objectMapper.isEnabled());
         assertEquals(ObjectMapper.Subobjects.AUTO, objectMapper.subobjects());
-        assertTrue(objectMapper.storeArraySource());
+        assertEquals(Mapper.SourceKeepMode.ARRAYS, objectMapper.sourceKeepMode().orElse(Mapper.SourceKeepMode.NONE));
     }
 
     public void testFieldReplacementForIndexTemplates() throws IOException {
@@ -678,14 +678,14 @@ public class ObjectMapperTests extends MapperServiceTestCase {
 
     public void testStoreArraySourceinSyntheticSourceMode() throws IOException {
         DocumentMapper mapper = createDocumentMapper(syntheticSourceMapping(b -> {
-            b.startObject("o").field("type", "object").field(ObjectMapper.STORE_ARRAY_SOURCE_PARAM, true).endObject();
+            b.startObject("o").field("type", "object").field("synthetic_source_keep", "arrays").endObject();
         }));
         assertNotNull(mapper.mapping().getRoot().getMapper("o"));
     }
 
     public void testStoreArraySourceNoopInNonSyntheticSourceMode() throws IOException {
         DocumentMapper mapper = createDocumentMapper(mapping(b -> {
-            b.startObject("o").field("type", "object").field(ObjectMapper.STORE_ARRAY_SOURCE_PARAM, true).endObject();
+            b.startObject("o").field("type", "object").field("synthetic_source_keep", "arrays").endObject();
         }));
         assertNotNull(mapper.mapping().getRoot().getMapper("o"));
     }
@@ -727,7 +727,7 @@ public class ObjectMapperTests extends MapperServiceTestCase {
                 b.field("subobjects", false);
                 b.field("enabled", false);
                 b.field("dynamic", false);
-                b.field(ObjectMapper.STORE_ARRAY_SOURCE_PARAM, true);
+                b.field("synthetic_source_keep", "arrays");
                 b.startObject("properties");
                 propertiesBuilder.accept(b);
                 b.endObject();

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

@@ -14,6 +14,7 @@ import org.elasticsearch.common.Explicit;
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
@@ -186,6 +187,7 @@ public class PassThroughObjectMapperTests extends MapperServiceTestCase {
             name,
             name,
             Explicit.EXPLICIT_TRUE,
+            Optional.empty(),
             ObjectMapper.Dynamic.FALSE,
             Map.of(),
             Explicit.EXPLICIT_FALSE,

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

@@ -362,6 +362,19 @@ public class RootObjectMapperTests extends MapperServiceTestCase {
         assertThat(e.getMessage(), containsString("type cannot be an empty string"));
     }
 
+    public void testSyntheticSourceKeepAllThrows() throws IOException {
+        String mapping = Strings.toString(
+            XContentFactory.jsonBuilder()
+                .startObject()
+                .startObject(MapperService.SINGLE_MAPPING_NAME)
+                .field("synthetic_source_keep", "all")
+                .endObject()
+                .endObject()
+        );
+        Exception e = expectThrows(MapperParsingException.class, () -> createMapperService(mapping));
+        assertThat(e.getMessage(), containsString("root object can't be configured with [synthetic_source_keep:all]"));
+    }
+
     public void testWithoutMappers() throws IOException {
         RootObjectMapper shallowRoot = createRootObjectMapperWithAllParametersSet(b -> {}, b -> {});
         RootObjectMapper root = createRootObjectMapperWithAllParametersSet(b -> {

test/framework/src/main/java/org/elasticsearch/index/mapper/MapperTestCase.java

@@ -1553,7 +1553,7 @@ public abstract class MapperTestCase extends MapperServiceTestCase {
         SyntheticSourceExample example = syntheticSourceSupportForKeepTests(shouldUseIgnoreMalformed()).example(1);
         DocumentMapper mapper = createDocumentMapper(syntheticSourceMapping(b -> {
             b.startObject("field");
-            b.field(Mapper.SYNTHETIC_SOURCE_KEEP_PARAM, "none");
+            b.field("synthetic_source_keep", "none");
             example.mapping().accept(b);
             b.endObject();
         }));
@@ -1564,7 +1564,7 @@ public abstract class MapperTestCase extends MapperServiceTestCase {
         SyntheticSourceExample example = syntheticSourceSupportForKeepTests(shouldUseIgnoreMalformed()).example(1);
         DocumentMapper mapperAll = createDocumentMapper(syntheticSourceMapping(b -> {
             b.startObject("field");
-            b.field(Mapper.SYNTHETIC_SOURCE_KEEP_PARAM, "all");
+            b.field("synthetic_source_keep", "all");
             example.mapping().accept(b);
             b.endObject();
         }));
@@ -1581,7 +1581,7 @@ public abstract class MapperTestCase extends MapperServiceTestCase {
         SyntheticSourceExample example = syntheticSourceSupportForKeepTests(shouldUseIgnoreMalformed()).example(1);
         DocumentMapper mapperAll = createDocumentMapper(syntheticSourceMapping(b -> {
             b.startObject("field");
-            b.field(Mapper.SYNTHETIC_SOURCE_KEEP_PARAM, randomFrom("arrays", "all"));  // Both options keep array source.
+            b.field("synthetic_source_keep", randomFrom("arrays", "all"));  // Both options keep array source.
             example.mapping().accept(b);
             b.endObject();
         }));

test/framework/src/main/java/org/elasticsearch/logsdb/datageneration/datasource/DataSourceRequest.java

@@ -12,6 +12,7 @@ package org.elasticsearch.logsdb.datageneration.datasource;
 import org.elasticsearch.logsdb.datageneration.DataGeneratorSpecification;
 import org.elasticsearch.logsdb.datageneration.FieldType;
 import org.elasticsearch.logsdb.datageneration.fields.DynamicMapping;
+import org.elasticsearch.test.ESTestCase;
 
 import java.util.Set;
 
@@ -115,11 +116,15 @@ public interface DataSourceRequest<TResponse extends DataSourceResponse> {
         }
     }
 
-    record ObjectMappingParametersGenerator(boolean isNested)
+    record ObjectMappingParametersGenerator(boolean isRoot, boolean isNested)
         implements
             DataSourceRequest<DataSourceResponse.ObjectMappingParametersGenerator> {
         public DataSourceResponse.ObjectMappingParametersGenerator accept(DataSourceHandler handler) {
             return handler.handle(this);
         }
+
+        public String syntheticSourceKeepValue() {
+            return isRoot() ? ESTestCase.randomFrom("none", "arrays") : ESTestCase.randomFrom("none", "arrays", "all");
+        }
     }
 }

test/framework/src/main/java/org/elasticsearch/logsdb/datageneration/datasource/DefaultMappingParametersHandler.java

@@ -83,6 +83,9 @@ public class DefaultMappingParametersHandler implements DataSourceHandler {
                 if (ESTestCase.randomBoolean()) {
                     parameters.put("dynamic", ESTestCase.randomFrom("true", "false", "strict"));
                 }
+                if (ESTestCase.randomBoolean()) {
+                    parameters.put(Mapper.SYNTHETIC_SOURCE_KEEP_PARAM, "all");  // [arrays] doesn't apply to nested objects
+                }
 
                 return parameters;
             });
@@ -96,6 +99,9 @@ public class DefaultMappingParametersHandler implements DataSourceHandler {
             if (ESTestCase.randomBoolean()) {
                 parameters.put("enabled", ESTestCase.randomFrom("true", "false"));
             }
+            if (ESTestCase.randomBoolean()) {
+                parameters.put(Mapper.SYNTHETIC_SOURCE_KEEP_PARAM, request.syntheticSourceKeepValue());
+            }
 
             return parameters;
         });

test/framework/src/main/java/org/elasticsearch/logsdb/datageneration/fields/NestedFieldDataGenerator.java

@@ -28,7 +28,7 @@ public class NestedFieldDataGenerator implements FieldDataGenerator {
 
         this.mappingParameters = context.specification()
             .dataSource()
-            .get(new DataSourceRequest.ObjectMappingParametersGenerator(true))
+            .get(new DataSourceRequest.ObjectMappingParametersGenerator(false, true))
             .mappingGenerator()
             .get();
         var dynamicMapping = context.determineDynamicMapping(mappingParameters);

test/framework/src/main/java/org/elasticsearch/logsdb/datageneration/fields/ObjectFieldDataGenerator.java

@@ -28,7 +28,7 @@ public class ObjectFieldDataGenerator implements FieldDataGenerator {
 
         this.mappingParameters = context.specification()
             .dataSource()
-            .get(new DataSourceRequest.ObjectMappingParametersGenerator(false))
+            .get(new DataSourceRequest.ObjectMappingParametersGenerator(false, false))
             .mappingGenerator()
             .get();
         var dynamicMapping = context.determineDynamicMapping(mappingParameters);

test/framework/src/main/java/org/elasticsearch/logsdb/datageneration/fields/TopLevelObjectFieldDataGenerator.java

@@ -37,7 +37,7 @@ public class TopLevelObjectFieldDataGenerator {
             this.mappingParameters = Map.of();
         } else {
             this.mappingParameters = new HashMap<>(
-                specification.dataSource().get(new DataSourceRequest.ObjectMappingParametersGenerator(false)).mappingGenerator().get()
+                specification.dataSource().get(new DataSourceRequest.ObjectMappingParametersGenerator(true, false)).mappingGenerator().get()
             );
             // Top-level object can't be disabled because @timestamp is a required field in data streams.
             this.mappingParameters.remove("enabled");

x-pack/plugin/otel-data/src/main/resources/component-templates/traces-otel@mappings.yaml

@@ -44,7 +44,7 @@ template:
       dropped_events_count:
         type: long
       links:
-        store_array_source: true
+        synthetic_source_keep: arrays
         properties:
           trace_id:
             type: keyword

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/security/authz_api_keys/40_document_level_security_synthetic_source.yml

@@ -186,7 +186,7 @@ Filter on object with stored source:
                 type: keyword
               obj:
                 type: object
-                store_array_source: true
+                synthetic_source_keep: arrays
                 properties:
                   secret:
                     type: keyword