Deprecate source mode in mappings (#117177)

Backport of #116689 to 8.18

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/rest/compat/compat/RestCompatTestTransformTask.java

@@ -135,8 +135,8 @@ public abstract class RestCompatTestTransformTask extends DefaultTask {
         // For example: indices.get_mapping/20_missing_type/Non-existent type returns 404
         // However, the folder can be arbitrarily nest so, a == a1/a2/a3, and the test name can include forward slashes, so c == c1/c2/c3
         // So we also need to support a1/a2/a3/b/c1/c2/c3
-
-        String[] testParts = fullTestName.split("/");
+        boolean limitTo3Separators = fullTestName.equals("logsdb/20_source_mapping/include/exclude is supported with stored _source");
+        String[] testParts = limitTo3Separators ? fullTestName.split("/", 3) : fullTestName.split("/");
         if (testParts.length < 3) {
             throw new IllegalArgumentException(
                 "To skip tests, all 3 parts [folder/file/test name] must be defined. found [" + fullTestName + "]"

docs/changelog/116689.yaml

@@ -0,0 +1,10 @@
+pr: 116689
+summary: Deprecate `_source.mode` in mappings
+area: Mapping
+type: deprecation
+issues: []
+deprecation:
+  title: Deprecate `_source.mode` in mappings
+  area: Mapping
+  details: Configuring `_source.mode` in mappings is deprecated and will be removed in future versions. Use `index.mapping.source.mode` index setting instead.
+  impact: Use `index.mapping.source.mode` index setting instead

modules/data-streams/src/test/java/org/elasticsearch/datastreams/mapper/DataStreamTimestampFieldMapperTests.java

@@ -48,7 +48,8 @@ public class DataStreamTimestampFieldMapperTests extends MetadataMapperTestCase
         checker.registerConflictCheck(
             "enabled",
             timestampMapping(true, b -> b.startObject("@timestamp").field("type", "date").endObject()),
-            timestampMapping(false, b -> b.startObject("@timestamp").field("type", "date").endObject())
+            timestampMapping(false, b -> b.startObject("@timestamp").field("type", "date").endObject()),
+            dm -> {}
         );
         checker.registerUpdateCheck(
             timestampMapping(false, b -> b.startObject("@timestamp").field("type", "date").endObject()),

qa/full-cluster-restart/src/javaRestTest/java/org/elasticsearch/upgrades/LogsIndexModeFullClusterRestartIT.java

@@ -17,7 +17,6 @@ import org.elasticsearch.client.RestClient;
 import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.time.FormatNames;
-import org.elasticsearch.test.MapMatcher;
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
 import org.elasticsearch.test.cluster.local.distribution.DistributionType;
 import org.elasticsearch.test.rest.RestTestLegacyFeatures;
@@ -31,9 +30,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
 
-import static org.elasticsearch.test.MapMatcher.assertMap;
-import static org.elasticsearch.test.MapMatcher.matchesMap;
-
 public class LogsIndexModeFullClusterRestartIT extends ParameterizedFullClusterRestartTestCase {
 
     @ClassRule
@@ -172,22 +168,16 @@ public class LogsIndexModeFullClusterRestartIT extends ParameterizedFullClusterR
             assertOK(bulkIndexResponse);
             assertThat(entityAsMap(bulkIndexResponse).get("errors"), Matchers.is(false));
 
-            assertIndexMappingsAndSettings(0, Matchers.nullValue(), matchesMap().extraOk());
-            assertIndexMappingsAndSettings(
-                1,
-                Matchers.equalTo("logsdb"),
-                matchesMap().extraOk().entry("_source", Map.of("mode", "synthetic"))
-            );
+            assertIndexSettings(0, Matchers.nullValue());
+            assertIndexSettings(1, Matchers.equalTo("logsdb"));
         }
     }
 
-    private void assertIndexMappingsAndSettings(int backingIndex, final Matcher<Object> indexModeMatcher, final MapMatcher mappingsMatcher)
-        throws IOException {
+    private void assertIndexSettings(int backingIndex, final Matcher<Object> indexModeMatcher) throws IOException {
         assertThat(
             getSettings(client(), getWriteBackingIndex(client(), "logs-apache-production", backingIndex)).get("index.mode"),
             indexModeMatcher
         );
-        assertMap(getIndexMappingAsMap(getWriteBackingIndex(client(), "logs-apache-production", backingIndex)), mappingsMatcher);
     }
 
     private static Request createDataStream(final String dataStreamName) {

qa/rolling-upgrade/src/javaRestTest/java/org/elasticsearch/upgrades/IndexingIT.java

@@ -18,6 +18,7 @@ import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.time.DateUtils;
 import org.elasticsearch.common.xcontent.support.XContentMapValues;
+import org.elasticsearch.index.IndexVersions;
 import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.index.mapper.TimeSeriesIdFieldMapper;
 import org.elasticsearch.test.ListMatcher;
@@ -419,9 +420,15 @@ public class IndexingIT extends AbstractRollingUpgradeTestCase {
         if (isOldCluster()) {
             Request createIndex = new Request("PUT", "/synthetic");
             XContentBuilder indexSpec = XContentBuilder.builder(XContentType.JSON.xContent()).startObject();
+            boolean useIndexSetting = getOldClusterIndexVersion().onOrAfter(IndexVersions.DEPRECATE_SOURCE_MODE_MAPPER);
+            if (useIndexSetting) {
+                indexSpec.startObject("settings").field("index.mapping.source.mode", "synthetic").endObject();
+            }
             indexSpec.startObject("mappings");
             {
-                indexSpec.startObject("_source").field("mode", "synthetic").endObject();
+                if (useIndexSetting == false) {
+                    indexSpec.startObject("_source").field("mode", "synthetic").endObject();
+                }
                 indexSpec.startObject("properties").startObject("kwd").field("type", "keyword").endObject().endObject();
             }
             indexSpec.endObject();

qa/rolling-upgrade/src/javaRestTest/java/org/elasticsearch/upgrades/LogsIndexModeRollingUpgradeIT.java

@@ -17,7 +17,6 @@ import org.elasticsearch.client.RestClient;
 import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.time.FormatNames;
-import org.elasticsearch.test.MapMatcher;
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
 import org.elasticsearch.test.cluster.local.distribution.DistributionType;
 import org.hamcrest.Matcher;
@@ -30,9 +29,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
 
-import static org.elasticsearch.test.MapMatcher.assertMap;
-import static org.elasticsearch.test.MapMatcher.matchesMap;
-
 public class LogsIndexModeRollingUpgradeIT extends AbstractRollingUpgradeTestCase {
 
     @ClassRule()
@@ -160,14 +156,10 @@ public class LogsIndexModeRollingUpgradeIT extends AbstractRollingUpgradeTestCas
             assertOK(bulkIndexResponse);
             assertThat(entityAsMap(bulkIndexResponse).get("errors"), Matchers.is(false));
 
-            assertIndexMappingsAndSettings(0, Matchers.nullValue(), matchesMap().extraOk());
-            assertIndexMappingsAndSettings(1, Matchers.nullValue(), matchesMap().extraOk());
-            assertIndexMappingsAndSettings(2, Matchers.nullValue(), matchesMap().extraOk());
-            assertIndexMappingsAndSettings(
-                3,
-                Matchers.equalTo("logsdb"),
-                matchesMap().extraOk().entry("_source", Map.of("mode", "synthetic"))
-            );
+            assertIndexSettings(0, Matchers.nullValue());
+            assertIndexSettings(1, Matchers.nullValue());
+            assertIndexSettings(2, Matchers.nullValue());
+            assertIndexSettings(3, Matchers.equalTo("logsdb"));
         }
     }
 
@@ -183,13 +175,11 @@ public class LogsIndexModeRollingUpgradeIT extends AbstractRollingUpgradeTestCas
         assertOK(client().performRequest(request));
     }
 
-    private void assertIndexMappingsAndSettings(int backingIndex, final Matcher<Object> indexModeMatcher, final MapMatcher mappingsMatcher)
-        throws IOException {
+    private void assertIndexSettings(int backingIndex, final Matcher<Object> indexModeMatcher) throws IOException {
         assertThat(
             getSettings(client(), getWriteBackingIndex(client(), "logs-apache-production", backingIndex)).get("index.mode"),
             indexModeMatcher
         );
-        assertMap(getIndexMappingAsMap(getWriteBackingIndex(client(), "logs-apache-production", backingIndex)), mappingsMatcher);
     }
 
     private static Request createDataStream(final String dataStreamName) {

rest-api-spec/build.gradle

@@ -247,4 +247,10 @@ tasks.named("precommit").configure {
 tasks.named("yamlRestTestV7CompatTransform").configure({ task ->
   task.skipTest("indices.sort/10_basic/Index Sort", "warning does not exist for compatibility")
   task.skipTest("search/330_fetch_fields/Test search rewrite", "warning does not exist for compatibility")
+  task.skipTest("tsdb/20_mapping/stored source is supported", "no longer serialize source_mode")
+  task.skipTest("tsdb/20_mapping/Synthetic source", "no longer serialize source_mode")
+  task.skipTest("logsdb/10_settings/create logs index", "no longer serialize source_mode")
+  task.skipTest("logsdb/20_source_mapping/stored _source mode is supported", "no longer serialize source_mode")
+  task.skipTest("logsdb/20_source_mapping/include/exclude is supported with stored _source", "no longer serialize source_mode")
+  task.skipTest("logsdb/20_source_mapping/synthetic _source is default", "no longer serialize source_mode")
 })

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/logsdb/10_settings.yml

@@ -77,11 +77,6 @@ create logs index:
   - is_true: test
   - match: { test.settings.index.mode: "logsdb" }
 
-  - do:
-      indices.get_mapping:
-        index: test
-  - match: { test.mappings._source.mode: synthetic }
-
 ---
 using default timestamp field mapping:
   - requires:

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/logsdb/20_source_mapping.yml

@@ -13,10 +13,10 @@ synthetic _source is default:
             index:
               mode: logsdb
   - do:
-      indices.get:
+      indices.get_settings:
         index: test-default-source
-
-  - match: { test-default-source.mappings._source.mode: "synthetic" }
+  - match: { test-default-source.settings.index.mode: logsdb }
+  - match: { test-default-source.settings.index.mapping.source.mode: null }
 
 ---
 stored _source mode is supported:
@@ -28,11 +28,12 @@ stored _source mode is supported:
             index:
               mode: logsdb
               mapping.source.mode: stored
+
   - do:
-      indices.get:
+      indices.get_settings:
         index: test-stored-source
-
-  - match: { test-stored-source.mappings._source.mode: "stored" }
+  - match: { test-stored-source.settings.index.mode: logsdb }
+  - match: { test-stored-source.settings.index.mapping.source.mode: stored }
 
 ---
 disabled _source is not supported:
@@ -110,7 +111,6 @@ include/exclude is supported with stored _source:
       indices.get:
         index: test-includes
 
-  - match: { test-includes.mappings._source.mode: "stored" }
   - match: { test-includes.mappings._source.includes: ["a"] }
 
   - do:
@@ -129,5 +129,4 @@ include/exclude is supported with stored _source:
       indices.get:
         index: test-excludes
 
-  - match: { test-excludes.mappings._source.mode: "stored" }
   - match: { test-excludes.mappings._source.excludes: ["b"] }

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/tsdb/20_mapping.yml

@@ -450,11 +450,6 @@ nested fields:
                 type: long
                 time_series_metric: gauge
 
-  - do:
-      indices.get_mapping: {}
-
-  - match: {tsdb-synthetic.mappings._source.mode: synthetic}
-
 ---
 stored source is supported:
   - requires:
@@ -486,12 +481,6 @@ stored source is supported:
                         type: keyword
                         time_series_dimension: true
 
-  - do:
-      indices.get:
-        index: tsdb_index
-
-  - match: { tsdb_index.mappings._source.mode: "stored" }
-
 ---
 disabled source is not supported:
   - requires:

server/src/main/java/org/elasticsearch/index/IndexSettingProvider.java

@@ -11,6 +11,7 @@ package org.elasticsearch.index;
 
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.CheckedFunction;
@@ -54,7 +55,7 @@ public interface IndexSettingProvider {
     /**
      * Infrastructure class that holds services that can be used by {@link IndexSettingProvider} instances.
      */
-    record Parameters(CheckedFunction<IndexMetadata, MapperService, IOException> mapperServiceFactory) {
+    record Parameters(ClusterService clusterService, CheckedFunction<IndexMetadata, MapperService, IOException> mapperServiceFactory) {
 
     }
 

server/src/main/java/org/elasticsearch/index/IndexVersions.java

@@ -121,6 +121,7 @@ public class IndexVersions {
     public static final IndexVersion ADD_ROLE_MAPPING_CLEANUP_MIGRATION = def(8_518_00_0, Version.LUCENE_9_12_0);
     public static final IndexVersion LOGSDB_DEFAULT_IGNORE_DYNAMIC_BEYOND_LIMIT_BACKPORT = def(8_519_00_0, Version.LUCENE_9_12_0);
     public static final IndexVersion TIME_BASED_K_ORDERED_DOC_ID_BACKPORT = def(8_520_00_0, Version.LUCENE_9_12_0);
+    public static final IndexVersion DEPRECATE_SOURCE_MODE_MAPPER = def(8_521_00_0, Version.LUCENE_9_12_0);
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

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

@@ -18,6 +18,7 @@ import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.common.Explicit;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.logging.DeprecationCategory;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.CollectionUtils;
@@ -38,6 +39,7 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 
 public class SourceFieldMapper extends MetadataFieldMapper {
     public static final NodeFeature SYNTHETIC_SOURCE_FALLBACK = new NodeFeature("mapper.source.synthetic_source_fallback");
@@ -68,6 +70,9 @@ public class SourceFieldMapper extends MetadataFieldMapper {
         return indexMode.defaultSourceMode().name();
     }, "index.mapping.source.mode", value -> {}, Setting.Property.Final, Setting.Property.IndexScope);
 
+    public static final String DEPRECATION_WARNING = "Configuring source mode in mappings is deprecated and will be removed "
+        + "in future versions. Use [index.mapping.source.mode] index setting instead.";
+
     /** The source mode */
     public enum Mode {
         DISABLED,
@@ -79,28 +84,32 @@ public class SourceFieldMapper extends MetadataFieldMapper {
         null,
         Explicit.IMPLICIT_TRUE,
         Strings.EMPTY_ARRAY,
-        Strings.EMPTY_ARRAY
+        Strings.EMPTY_ARRAY,
+        false
     );
 
     private static final SourceFieldMapper STORED = new SourceFieldMapper(
         Mode.STORED,
         Explicit.IMPLICIT_TRUE,
         Strings.EMPTY_ARRAY,
-        Strings.EMPTY_ARRAY
+        Strings.EMPTY_ARRAY,
+        false
     );
 
     private static final SourceFieldMapper SYNTHETIC = new SourceFieldMapper(
         Mode.SYNTHETIC,
         Explicit.IMPLICIT_TRUE,
         Strings.EMPTY_ARRAY,
-        Strings.EMPTY_ARRAY
+        Strings.EMPTY_ARRAY,
+        false
     );
 
     private static final SourceFieldMapper DISABLED = new SourceFieldMapper(
         Mode.DISABLED,
         Explicit.IMPLICIT_TRUE,
         Strings.EMPTY_ARRAY,
-        Strings.EMPTY_ARRAY
+        Strings.EMPTY_ARRAY,
+        false
     );
 
     public static class Defaults {
@@ -134,16 +143,7 @@ public class SourceFieldMapper extends MetadataFieldMapper {
          * The default mode for TimeSeries is left empty on purpose, so that mapping printings include the synthetic
          * source mode.
          */
-        private final Parameter<Mode> mode = new Parameter<>(
-            "mode",
-            true,
-            () -> null,
-            (n, c, o) -> Mode.valueOf(o.toString().toUpperCase(Locale.ROOT)),
-            m -> toType(m).enabled.explicit() ? null : toType(m).mode,
-            (b, n, v) -> b.field(n, v.toString().toLowerCase(Locale.ROOT)),
-            v -> v.toString().toLowerCase(Locale.ROOT)
-        ).setMergeValidator((previous, current, conflicts) -> (previous == current) || current != Mode.STORED)
-            .setSerializerCheck((includeDefaults, isConfigured, value) -> value != null); // don't emit if `enabled` is configured
+        private final Parameter<Mode> mode;
         private final Parameter<List<String>> includes = Parameter.stringArrayParam(
             "includes",
             false,
@@ -158,15 +158,28 @@ public class SourceFieldMapper extends MetadataFieldMapper {
         private final Settings settings;
 
         private final IndexMode indexMode;
+        private boolean serializeMode;
 
         private final boolean supportsNonDefaultParameterValues;
 
-        public Builder(IndexMode indexMode, final Settings settings, boolean supportsCheckForNonDefaultParams) {
+        public Builder(IndexMode indexMode, final Settings settings, boolean supportsCheckForNonDefaultParams, boolean serializeMode) {
             super(Defaults.NAME);
             this.settings = settings;
             this.indexMode = indexMode;
             this.supportsNonDefaultParameterValues = supportsCheckForNonDefaultParams == false
                 || settings.getAsBoolean(LOSSY_PARAMETERS_ALLOWED_SETTING_NAME, true);
+            this.serializeMode = serializeMode;
+            this.mode = new Parameter<>(
+                "mode",
+                true,
+                () -> null,
+                (n, c, o) -> Mode.valueOf(o.toString().toUpperCase(Locale.ROOT)),
+                m -> toType(m).enabled.explicit() ? null : toType(m).mode,
+                (b, n, v) -> b.field(n, v.toString().toLowerCase(Locale.ROOT)),
+                v -> v.toString().toLowerCase(Locale.ROOT)
+            ).setMergeValidator((previous, current, conflicts) -> (previous == current) || current != Mode.STORED)
+                // don't emit if `enabled` is configured
+                .setSerializerCheck((includeDefaults, isConfigured, value) -> serializeMode && value != null);
         }
 
         public Builder setSynthetic() {
@@ -219,21 +232,22 @@ public class SourceFieldMapper extends MetadataFieldMapper {
             if (sourceMode == Mode.SYNTHETIC && (includes.getValue().isEmpty() == false || excludes.getValue().isEmpty() == false)) {
                 throw new IllegalArgumentException("filtering the stored _source is incompatible with synthetic source");
             }
-
-            SourceFieldMapper sourceFieldMapper;
-            if (isDefault()) {
+            if (mode.isConfigured()) {
+                serializeMode = true;
+            }
+            final SourceFieldMapper sourceFieldMapper;
+            if (isDefault() && sourceMode == null) {
                 // Needed for bwc so that "mode" is not serialized in case of a standard index with stored source.
-                if (sourceMode == null) {
-                    sourceFieldMapper = DEFAULT;
-                } else {
-                    sourceFieldMapper = resolveStaticInstance(sourceMode);
-                }
+                sourceFieldMapper = DEFAULT;
+            } else if (isDefault() && serializeMode == false && sourceMode != null) {
+                sourceFieldMapper = resolveStaticInstance(sourceMode);
             } else {
                 sourceFieldMapper = new SourceFieldMapper(
                     sourceMode,
                     enabled.get(),
                     includes.getValue().toArray(Strings.EMPTY_ARRAY),
-                    excludes.getValue().toArray(Strings.EMPTY_ARRAY)
+                    excludes.getValue().toArray(Strings.EMPTY_ARRAY),
+                    serializeMode
                 );
             }
             if (indexMode != null) {
@@ -283,15 +297,29 @@ public class SourceFieldMapper extends MetadataFieldMapper {
         if (indexMode == IndexMode.STANDARD && settingSourceMode == Mode.STORED) {
             return DEFAULT;
         }
-
-        return resolveStaticInstance(settingSourceMode);
+        if (c.indexVersionCreated().onOrAfter(IndexVersions.DEPRECATE_SOURCE_MODE_MAPPER)) {
+            return resolveStaticInstance(settingSourceMode);
+        } else {
+            return new SourceFieldMapper(settingSourceMode, Explicit.IMPLICIT_TRUE, Strings.EMPTY_ARRAY, Strings.EMPTY_ARRAY, true);
+        }
     },
         c -> new Builder(
             c.getIndexSettings().getMode(),
             c.getSettings(),
-            c.indexVersionCreated().onOrAfter(IndexVersions.SOURCE_MAPPER_LOSSY_PARAMS_CHECK)
+            c.indexVersionCreated().onOrAfter(IndexVersions.SOURCE_MAPPER_LOSSY_PARAMS_CHECK),
+            c.indexVersionCreated().before(IndexVersions.DEPRECATE_SOURCE_MODE_MAPPER)
         )
-    );
+    ) {
+        @Override
+        public MetadataFieldMapper.Builder parse(String name, Map<String, Object> node, MappingParserContext parserContext)
+            throws MapperParsingException {
+            assert name.equals(SourceFieldMapper.NAME) : name;
+            if (parserContext.indexVersionCreated().onOrAfter(IndexVersions.DEPRECATE_SOURCE_MODE_MAPPER) && node.containsKey("mode")) {
+                deprecationLogger.critical(DeprecationCategory.MAPPINGS, "mapping_source_mode", SourceFieldMapper.DEPRECATION_WARNING);
+            }
+            return super.parse(name, node, parserContext);
+        }
+    };
 
     static final class SourceFieldType extends MappedFieldType {
         private final boolean enabled;
@@ -330,8 +358,9 @@ public class SourceFieldMapper extends MetadataFieldMapper {
         }
     }
 
-    // nullable for bwc reasons
+    // nullable for bwc reasons - TODO: fold this into serializeMode
     private final @Nullable Mode mode;
+    private final boolean serializeMode;
     private final Explicit<Boolean> enabled;
 
     /** indicates whether the source will always exist and be complete, for use by features like the update API */
@@ -341,7 +370,7 @@ public class SourceFieldMapper extends MetadataFieldMapper {
     private final String[] excludes;
     private final SourceFilter sourceFilter;
 
-    private SourceFieldMapper(Mode mode, Explicit<Boolean> enabled, String[] includes, String[] excludes) {
+    private SourceFieldMapper(Mode mode, Explicit<Boolean> enabled, String[] includes, String[] excludes, boolean serializeMode) {
         super(new SourceFieldType((enabled.explicit() && enabled.value()) || (enabled.explicit() == false && mode != Mode.DISABLED)));
         this.mode = mode;
         this.enabled = enabled;
@@ -349,6 +378,7 @@ public class SourceFieldMapper extends MetadataFieldMapper {
         this.includes = includes;
         this.excludes = excludes;
         this.complete = stored() && sourceFilter == null;
+        this.serializeMode = serializeMode;
     }
 
     private static SourceFilter buildSourceFilter(String[] includes, String[] excludes) {
@@ -419,7 +449,7 @@ public class SourceFieldMapper extends MetadataFieldMapper {
 
     @Override
     public FieldMapper.Builder getMergeBuilder() {
-        return new Builder(null, Settings.EMPTY, false).init(this);
+        return new Builder(null, Settings.EMPTY, false, serializeMode).init(this);
     }
 
     /**

server/src/main/java/org/elasticsearch/node/NodeConstruction.java

@@ -827,7 +827,7 @@ class NodeConstruction {
             .searchOperationListeners(searchOperationListeners)
             .build();
 
-        final var parameters = new IndexSettingProvider.Parameters(indicesService::createIndexMapperServiceForValidation);
+        final var parameters = new IndexSettingProvider.Parameters(clusterService, indicesService::createIndexMapperServiceForValidation);
         IndexSettingProviders indexSettingProviders = new IndexSettingProviders(
             Sets.union(
                 builtinIndexSettingProviders(),

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

@@ -133,5 +133,6 @@ public class DocumentParserContextTests extends ESTestCase {
         assertEquals(ObjectMapper.Defaults.DYNAMIC, resultFromParserContext.getDynamic());
         assertEquals(MapperService.MergeReason.MAPPING_UPDATE, resultFromParserContext.getMergeReason());
         assertFalse(resultFromParserContext.isInNestedContext());
+        assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
     }
 }

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

@@ -69,7 +69,7 @@ public class DynamicFieldsBuilderTests extends ESTestCase {
         XContentParser parser = createParser(JsonXContent.jsonXContent, source);
         SourceToParse sourceToParse = new SourceToParse("test", new BytesArray(source), XContentType.JSON);
 
-        SourceFieldMapper sourceMapper = new SourceFieldMapper.Builder(null, Settings.EMPTY, false).setSynthetic().build();
+        SourceFieldMapper sourceMapper = new SourceFieldMapper.Builder(null, Settings.EMPTY, false, false).setSynthetic().build();
         RootObjectMapper root = new RootObjectMapper.Builder("_doc", Optional.empty()).add(
             new PassThroughObjectMapper.Builder("labels").setPriority(0).setContainsDimensions().dynamic(ObjectMapper.Dynamic.TRUE)
         ).build(MapperBuilderContext.root(false, false));

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

@@ -52,7 +52,8 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
         checker.registerConflictCheck(
             "enabled",
             topMapping(b -> b.startObject(SourceFieldMapper.NAME).field("enabled", false).endObject()),
-            topMapping(b -> b.startObject(SourceFieldMapper.NAME).field("enabled", true).endObject())
+            topMapping(b -> b.startObject(SourceFieldMapper.NAME).field("enabled", true).endObject()),
+            dm -> {}
         );
         checker.registerUpdateCheck(
             topMapping(b -> b.startObject(SourceFieldMapper.NAME).field("enabled", true).endObject()),
@@ -62,14 +63,18 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
         checker.registerUpdateCheck(
             topMapping(b -> b.startObject(SourceFieldMapper.NAME).field("mode", "stored").endObject()),
             topMapping(b -> b.startObject(SourceFieldMapper.NAME).field("mode", "synthetic").endObject()),
-            dm -> assertTrue(dm.metadataMapper(SourceFieldMapper.class).isSynthetic())
+            dm -> {
+                assertTrue(dm.metadataMapper(SourceFieldMapper.class).isSynthetic());
+                assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
+            }
         );
         checker.registerConflictCheck("includes", b -> b.array("includes", "foo*"));
         checker.registerConflictCheck("excludes", b -> b.array("excludes", "foo*"));
         checker.registerConflictCheck(
             "mode",
             topMapping(b -> b.startObject(SourceFieldMapper.NAME).field("mode", "synthetic").endObject()),
-            topMapping(b -> b.startObject(SourceFieldMapper.NAME).field("mode", "stored").endObject())
+            topMapping(b -> b.startObject(SourceFieldMapper.NAME).field("mode", "stored").endObject()),
+            dm -> assertWarnings(SourceFieldMapper.DEPRECATION_WARNING)
         );
     }
 
@@ -206,13 +211,14 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
             )
         );
         assertThat(e.getMessage(), containsString("Cannot set both [mode] and [enabled] parameters"));
+        assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
     }
 
     public void testSyntheticUpdates() throws Exception {
         MapperService mapperService = createMapperService("""
             { "_doc" : { "_source" : { "mode" : "synthetic" } } }
             """);
-
+        assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
         SourceFieldMapper mapper = mapperService.documentMapper().sourceMapper();
         assertTrue(mapper.enabled());
         assertTrue(mapper.isSynthetic());
@@ -220,6 +226,7 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
         merge(mapperService, """
             { "_doc" : { "_source" : { "mode" : "synthetic" } } }
             """);
+        assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
         mapper = mapperService.documentMapper().sourceMapper();
         assertTrue(mapper.enabled());
         assertTrue(mapper.isSynthetic());
@@ -230,11 +237,15 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
         Exception e = expectThrows(IllegalArgumentException.class, () -> merge(mapperService, """
             { "_doc" : { "_source" : { "mode" : "stored" } } }
             """));
+
         assertThat(e.getMessage(), containsString("Cannot update parameter [mode] from [synthetic] to [stored]"));
+        assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
 
         merge(mapperService, """
             { "_doc" : { "_source" : { "mode" : "disabled" } } }
             """);
+        assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
+
         mapper = mapperService.documentMapper().sourceMapper();
         assertFalse(mapper.enabled());
         assertFalse(mapper.isSynthetic());
@@ -247,14 +258,14 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
         });
         DocumentMapper mapper = createTimeSeriesModeDocumentMapper(mapping);
         assertTrue(mapper.sourceMapper().isSynthetic());
-        assertEquals("{\"_source\":{\"mode\":\"synthetic\"}}", mapper.sourceMapper().toString());
+        assertEquals("{\"_source\":{}}", mapper.sourceMapper().toString());
     }
 
     public void testSyntheticSourceWithLogsIndexMode() throws IOException {
         XContentBuilder mapping = fieldMapping(b -> { b.field("type", "keyword"); });
         DocumentMapper mapper = createLogsModeDocumentMapper(mapping);
         assertTrue(mapper.sourceMapper().isSynthetic());
-        assertEquals("{\"_source\":{\"mode\":\"synthetic\"}}", mapper.sourceMapper().toString());
+        assertEquals("{\"_source\":{}}", mapper.sourceMapper().toString());
     }
 
     public void testSupportsNonDefaultParameterValues() throws IOException {
@@ -270,6 +281,7 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
                 topMapping(b -> b.startObject("_source").field("mode", randomBoolean() ? "synthetic" : "stored").endObject())
             ).documentMapper().sourceMapper();
             assertThat(sourceFieldMapper, notNullValue());
+            assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
         }
         Exception e = expectThrows(
             MapperParsingException.class,
@@ -301,6 +313,8 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
                 .documentMapper()
                 .sourceMapper()
         );
+        assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
+
         assertThat(e.getMessage(), containsString("Parameter [mode=disabled] is not allowed in source"));
 
         e = expectThrows(
@@ -409,6 +423,7 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
             ParsedDocument doc = docMapper.parse(source(b -> { b.field("field1", "value1"); }));
             assertNotNull(doc.rootDoc().getField("_recovery_source"));
             assertThat(doc.rootDoc().getField("_recovery_source").binaryValue(), equalTo(new BytesRef("{\"field1\":\"value1\"}")));
+            assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
         }
         {
             Settings settings = Settings.builder().put(INDICES_RECOVERY_SOURCE_ENABLED_SETTING.getKey(), false).build();
@@ -419,6 +434,7 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
             DocumentMapper docMapper = mapperService.documentMapper();
             ParsedDocument doc = docMapper.parse(source(b -> b.field("field1", "value1")));
             assertNull(doc.rootDoc().getField("_recovery_source"));
+            assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
         }
     }
 
@@ -613,6 +629,7 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
             ParsedDocument doc = docMapper.parse(source(b -> { b.field("@timestamp", "2012-02-13"); }));
             assertNotNull(doc.rootDoc().getField("_recovery_source"));
             assertThat(doc.rootDoc().getField("_recovery_source").binaryValue(), equalTo(new BytesRef("{\"@timestamp\":\"2012-02-13\"}")));
+            assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
         }
         {
             Settings settings = Settings.builder()
@@ -623,6 +640,7 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
             DocumentMapper docMapper = mapperService.documentMapper();
             ParsedDocument doc = docMapper.parse(source(b -> b.field("@timestamp", "2012-02-13")));
             assertNull(doc.rootDoc().getField("_recovery_source"));
+            assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
         }
     }
 
@@ -691,6 +709,7 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
                 doc.rootDoc().getField("_recovery_source").binaryValue(),
                 equalTo(new BytesRef("{\"@timestamp\":\"2012-02-13\",\"field\":\"value1\"}"))
             );
+            assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
         }
         {
             Settings settings = Settings.builder()
@@ -704,6 +723,7 @@ public class SourceFieldMapperTests extends MetadataMapperTestCase {
                 source("123", b -> b.field("@timestamp", "2012-02-13").field("field", randomAlphaOfLength(5)), null)
             );
             assertNull(doc.rootDoc().getField("_recovery_source"));
+            assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
         }
     }
 }

server/src/test/java/org/elasticsearch/index/query/SearchExecutionContextTests.java

@@ -384,7 +384,7 @@ public class SearchExecutionContextTests extends ESTestCase {
 
     public void testSyntheticSourceSearchLookup() throws IOException {
         // Build a mapping using synthetic source
-        SourceFieldMapper sourceMapper = new SourceFieldMapper.Builder(null, Settings.EMPTY, false).setSynthetic().build();
+        SourceFieldMapper sourceMapper = new SourceFieldMapper.Builder(null, Settings.EMPTY, false, false).setSynthetic().build();
         RootObjectMapper root = new RootObjectMapper.Builder("_doc", Optional.empty()).add(
             new KeywordFieldMapper.Builder("cat", IndexVersion.current()).ignoreAbove(100)
         ).build(MapperBuilderContext.root(true, false));

server/src/test/java/org/elasticsearch/index/shard/ShardGetServiceTests.java

@@ -21,6 +21,7 @@ import org.elasticsearch.index.engine.LiveVersionMapTestUtils;
 import org.elasticsearch.index.engine.VersionConflictEngineException;
 import org.elasticsearch.index.get.GetResult;
 import org.elasticsearch.index.mapper.RoutingFieldMapper;
+import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 import org.elasticsearch.xcontent.XContentType;
 
@@ -114,6 +115,7 @@ public class ShardGetServiceTests extends IndexShardTestCase {
             "mode": "synthetic"
             """;
         runGetFromTranslogWithOptions(docToIndex, sourceOptions, expectedFetchedSource, "\"long\"", 7L, true);
+        assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
     }
 
     public void testGetFromTranslogWithDenseVector() throws IOException {

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

@@ -39,7 +39,7 @@ public abstract class MetadataMapperTestCase extends MapperServiceTestCase {
 
     protected abstract void registerParameters(ParameterChecker checker) throws IOException;
 
-    private record ConflictCheck(XContentBuilder init, XContentBuilder update) {}
+    private record ConflictCheck(XContentBuilder init, XContentBuilder update, Consumer<DocumentMapper> check) {}
 
     private record UpdateCheck(XContentBuilder init, XContentBuilder update, Consumer<DocumentMapper> check) {}
 
@@ -59,7 +59,7 @@ public abstract class MetadataMapperTestCase extends MapperServiceTestCase {
                 b.startObject(fieldName());
                 update.accept(b);
                 b.endObject();
-            })));
+            }), d -> {}));
         }
 
         /**
@@ -69,8 +69,8 @@ public abstract class MetadataMapperTestCase extends MapperServiceTestCase {
          * @param init   the initial mapping
          * @param update the updated mapping
          */
-        public void registerConflictCheck(String param, XContentBuilder init, XContentBuilder update) {
-            conflictChecks.put(param, new ConflictCheck(init, update));
+        public void registerConflictCheck(String param, XContentBuilder init, XContentBuilder update, Consumer<DocumentMapper> check) {
+            conflictChecks.put(param, new ConflictCheck(init, update, check));
         }
 
         public void registerUpdateCheck(XContentBuilder init, XContentBuilder update, Consumer<DocumentMapper> check) {
@@ -96,6 +96,7 @@ public abstract class MetadataMapperTestCase extends MapperServiceTestCase {
                 e.getMessage(),
                 anyOf(containsString("Cannot update parameter [" + param + "]"), containsString("different [" + param + "]"))
             );
+            checker.conflictChecks.get(param).check.accept(mapperService.documentMapper());
         }
         for (UpdateCheck updateCheck : checker.updateChecks) {
             MapperService mapperService = createMapperService(updateCheck.init);

test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java

@@ -68,6 +68,7 @@ import org.elasticsearch.health.node.selection.HealthNode;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.IndexVersions;
+import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.seqno.ReplicationTracker;
 import org.elasticsearch.rest.RestStatus;
@@ -1883,8 +1884,10 @@ public abstract class ESRestTestCase extends ESTestCase {
 
         if (settings != null && settings.getAsBoolean(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true) == false) {
             expectSoftDeletesWarning(request, name);
-        }
-
+        } else if (isSyntheticSourceConfiguredInMapping(mapping)
+            && minimumIndexVersion().onOrAfter(IndexVersions.DEPRECATE_SOURCE_MODE_MAPPER)) {
+                request.setOptions(expectVersionSpecificWarnings(v -> v.current(SourceFieldMapper.DEPRECATION_WARNING)));
+            }
         final Response response = client.performRequest(request);
         try (var parser = responseAsParser(response)) {
             return CreateIndexResponse.fromXContent(parser);
@@ -1928,6 +1931,49 @@ public abstract class ESRestTestCase extends ESTestCase {
         }));
     }
 
+    @SuppressWarnings("unchecked")
+    protected static boolean isSyntheticSourceConfiguredInMapping(String mapping) {
+        if (mapping == null) {
+            return false;
+        }
+        var mappings = XContentHelper.convertToMap(
+            JsonXContent.jsonXContent,
+            mapping.trim().startsWith("{") ? mapping : '{' + mapping + '}',
+            false
+        );
+        if (mappings.containsKey("_doc")) {
+            mappings = (Map<String, Object>) mappings.get("_doc");
+        }
+        Map<String, Object> sourceMapper = (Map<String, Object>) mappings.get(SourceFieldMapper.NAME);
+        if (sourceMapper == null) {
+            return false;
+        }
+        return sourceMapper.get("mode") != null;
+    }
+
+    @SuppressWarnings("unchecked")
+    protected static boolean isSyntheticSourceConfiguredInTemplate(String template) {
+        if (template == null) {
+            return false;
+        }
+        var values = XContentHelper.convertToMap(JsonXContent.jsonXContent, template, false);
+        for (Object value : values.values()) {
+            Map<String, Object> mappings = (Map<String, Object>) ((Map<String, Object>) value).get("mappings");
+            if (mappings == null) {
+                continue;
+            }
+            Map<String, Object> sourceMapper = (Map<String, Object>) mappings.get(SourceFieldMapper.NAME);
+            if (sourceMapper == null) {
+                continue;
+            }
+            Object mode = sourceMapper.get("mode");
+            if (mode != null) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     protected static Map<String, Object> getIndexSettings(String index) throws IOException {
         Request request = new Request("GET", "/" + index + "/_settings");
         request.addParameter("flat_settings", "true");
@@ -2321,7 +2367,7 @@ public abstract class ESRestTestCase extends ESTestCase {
      */
     protected static IndexVersion minimumIndexVersion() throws IOException {
         final Request request = new Request("GET", "_nodes");
-        request.addParameter("filter_path", "nodes.*.version,nodes.*.max_index_version");
+        request.addParameter("filter_path", "nodes.*.version,nodes.*.max_index_version,nodes.*.index_version");
 
         final Response response = adminClient().performRequest(request);
         final Map<String, Object> nodes = ObjectPath.createFromResponse(response).evaluate("nodes");
@@ -2329,10 +2375,13 @@ public abstract class ESRestTestCase extends ESTestCase {
         IndexVersion minVersion = null;
         for (Map.Entry<String, Object> node : nodes.entrySet()) {
             Map<?, ?> nodeData = (Map<?, ?>) node.getValue();
-            String versionStr = (String) nodeData.get("max_index_version");
+            Object versionStr = nodeData.get("index_version");
+            if (versionStr == null) {
+                versionStr = nodeData.get("max_index_version");
+            }
             // fallback on version if index version is not there
             IndexVersion indexVersion = versionStr != null
-                ? IndexVersion.fromId(Integer.parseInt(versionStr))
+                ? IndexVersion.fromId(Integer.parseInt(versionStr.toString()))
                 : IndexVersion.fromId(
                     parseLegacyVersion((String) nodeData.get("version")).map(Version::id).orElse(IndexVersions.MINIMUM_COMPATIBLE.id())
                 );

test/yaml-rest-runner/src/main/java/org/elasticsearch/test/rest/yaml/section/DoSection.java

@@ -20,6 +20,7 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.logging.HeaderWarning;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.core.UpdateForV9;
+import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.rest.action.admin.indices.RestPutIndexTemplateAction;
 import org.elasticsearch.test.rest.RestTestLegacyFeatures;
 import org.elasticsearch.test.rest.yaml.ClientYamlTestExecutionContext;
@@ -506,6 +507,8 @@ public class DoSection implements ExecutableSection {
             }
         }
 
+        unexpected.removeIf(s -> s.endsWith(SourceFieldMapper.DEPRECATION_WARNING + "\""));
+
         if (unexpected.isEmpty() == false
             || unmatched.isEmpty() == false
             || missing.isEmpty() == false

x-pack/plugin/ccr/qa/multi-cluster/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexIT.java

@@ -18,6 +18,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.mapper.DateFieldMapper;
+import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.repositories.fs.FsRepository;
 import org.elasticsearch.rest.RestStatus;
 
@@ -366,8 +367,10 @@ public class FollowIndexIT extends ESCCRRestTestCase {
         final String leaderIndexName = "synthetic_leader";
         if ("leader".equals(targetCluster)) {
             logger.info("Running against leader cluster");
-            createIndex(adminClient(), leaderIndexName, Settings.EMPTY, """
-                "_source": {"mode": "synthetic"},
+            Settings settings = Settings.builder()
+                .put(SourceFieldMapper.INDEX_MAPPER_SOURCE_MODE_SETTING.getKey(), SourceFieldMapper.Mode.SYNTHETIC)
+                .build();
+            createIndex(adminClient(), leaderIndexName, settings, """
                 "properties": {"kwd": {"type": "keyword"}}}""", null);
             for (int i = 0; i < numDocs; i++) {
                 logger.info("Indexing doc [{}]", i);
@@ -392,7 +395,6 @@ public class FollowIndexIT extends ESCCRRestTestCase {
             }
             assertBusy(() -> {
                 verifyDocuments(client(), followIndexName, numDocs);
-                assertMap(getIndexMappingAsMap(followIndexName), matchesMap().extraOk().entry("_source", Map.of("mode", "synthetic")));
                 if (overrideNumberOfReplicas) {
                     assertMap(getIndexSettingsAsMap(followIndexName), matchesMap().extraOk().entry("index.number_of_replicas", "0"));
                 } else {

x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/DeprecationChecks.java

@@ -97,7 +97,8 @@ public class DeprecationChecks {
         IndexDeprecationChecks::checkIndexDataPath,
         IndexDeprecationChecks::storeTypeSettingCheck,
         IndexDeprecationChecks::frozenIndexSettingCheck,
-        IndexDeprecationChecks::deprecatedCamelCasePattern
+        IndexDeprecationChecks::deprecatedCamelCasePattern,
+        IndexDeprecationChecks::checkSourceModeInMapping
     );
 
     static List<BiFunction<DataStream, ClusterState, DeprecationIssue>> DATA_STREAM_CHECKS = List.of(

x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/IndexDeprecationChecks.java

@@ -16,6 +16,7 @@ import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.IndexVersions;
 import org.elasticsearch.index.engine.frozen.FrozenEngine;
+import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.xpack.core.deprecation.DeprecationIssue;
 
 import java.util.ArrayList;
@@ -201,6 +202,31 @@ public class IndexDeprecationChecks {
         return issues;
     }
 
+    static DeprecationIssue checkSourceModeInMapping(IndexMetadata indexMetadata, ClusterState clusterState) {
+        if (indexMetadata.getCreationVersion().onOrAfter(IndexVersions.DEPRECATE_SOURCE_MODE_MAPPER)) {
+            boolean[] useSourceMode = { false };
+            fieldLevelMappingIssue(indexMetadata, ((mappingMetadata, sourceAsMap) -> {
+                Object source = sourceAsMap.get("_source");
+                if (source instanceof Map<?, ?> sourceMap) {
+                    if (sourceMap.containsKey("mode")) {
+                        useSourceMode[0] = true;
+                    }
+                }
+            }));
+            if (useSourceMode[0]) {
+                return new DeprecationIssue(
+                    DeprecationIssue.Level.CRITICAL,
+                    SourceFieldMapper.DEPRECATION_WARNING,
+                    "https://github.com/elastic/elasticsearch/pull/117172",
+                    SourceFieldMapper.DEPRECATION_WARNING,
+                    false,
+                    null
+                );
+            }
+        }
+        return null;
+    }
+
     static DeprecationIssue deprecatedCamelCasePattern(IndexMetadata indexMetadata, ClusterState clusterState) {
         List<String> fields = new ArrayList<>();
         fieldLevelMappingIssue(

x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/FieldExtractorTestCase.java

@@ -1775,16 +1775,12 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
     }
 
     private static void createIndex(String name, CheckedConsumer<XContentBuilder, IOException> mapping) throws IOException {
-        Request request = new Request("PUT", "/" + name);
         XContentBuilder index = JsonXContent.contentBuilder().prettyPrint().startObject();
-        index.startObject("mappings");
         mapping.accept(index);
         index.endObject();
-        index.endObject();
         String configStr = Strings.toString(index);
         logger.info("index: {} {}", name, configStr);
-        request.setJsonEntity(configStr);
-        client().performRequest(request);
+        ESRestTestCase.createIndex(name, Settings.EMPTY, configStr);
     }
 
     /**

x-pack/plugin/logsdb/src/javaRestTest/java/org/elasticsearch/xpack/logsdb/LogsIndexModeCustomSettingsIT.java

@@ -7,9 +7,12 @@
 
 package org.elasticsearch.xpack.logsdb;
 
+import org.elasticsearch.client.Request;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.IndexVersions;
+import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
 import org.elasticsearch.test.cluster.local.distribution.DistributionType;
 import org.hamcrest.Matchers;
@@ -113,8 +116,11 @@ public class LogsIndexModeCustomSettingsIT extends LogsIndexModeRestTestIT {
             }""";
 
         assertOK(putComponentTemplate(client, "logs@custom", storedSourceMapping));
-        assertOK(createDataStream(client, "logs-custom-dev"));
-
+        Request request = new Request("PUT", "_data_stream/logs-custom-dev");
+        if (minimumIndexVersion().onOrAfter(IndexVersions.DEPRECATE_SOURCE_MODE_MAPPER)) {
+            request.setOptions(expectVersionSpecificWarnings(v -> v.current(SourceFieldMapper.DEPRECATION_WARNING)));
+        }
+        assertOK(client.performRequest(request));
         var mapping = getMapping(client, getDataStreamBackingIndex(client, "logs-custom-dev", 0));
         String sourceMode = (String) subObject("_source").apply(mapping).get("mode");
         assertThat(sourceMode, equalTo("stored"));
@@ -183,7 +189,11 @@ public class LogsIndexModeCustomSettingsIT extends LogsIndexModeRestTestIT {
             }""";
 
         assertOK(putComponentTemplate(client, "logs@custom", storedSourceMapping));
-        assertOK(createDataStream(client, "logs-custom-dev"));
+        Request request = new Request("PUT", "_data_stream/logs-custom-dev");
+        if (minimumIndexVersion().onOrAfter(IndexVersions.DEPRECATE_SOURCE_MODE_MAPPER)) {
+            request.setOptions(expectVersionSpecificWarnings(v -> v.current(SourceFieldMapper.DEPRECATION_WARNING)));
+        }
+        assertOK(client.performRequest(request));
 
         var mapping = getMapping(client, getDataStreamBackingIndex(client, "logs-custom-dev", 0));
         String sourceMode = (String) subObject("_source").apply(mapping).get("mode");

x-pack/plugin/logsdb/src/javaRestTest/java/org/elasticsearch/xpack/logsdb/LogsIndexModeRestTestIT.java

@@ -11,6 +11,8 @@ import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.client.RestClient;
+import org.elasticsearch.index.IndexVersions;
+import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.test.rest.ESRestTestCase;
 
 import java.io.IOException;
@@ -35,6 +37,12 @@ public abstract class LogsIndexModeRestTestIT extends ESRestTestCase {
         throws IOException {
         final Request request = new Request("PUT", "/_component_template/" + componentTemplate);
         request.setJsonEntity(contends);
+        if (isSyntheticSourceConfiguredInTemplate(contends)
+            && minimumIndexVersion().onOrAfter(IndexVersions.DEPRECATE_SOURCE_MODE_MAPPER)) {
+            request.setOptions(
+                expectVersionSpecificWarnings((VersionSensitiveWarningsHandler v) -> v.current(SourceFieldMapper.DEPRECATION_WARNING))
+            );
+        }
         return client.performRequest(request);
     }
 

x-pack/plugin/logsdb/src/main/java/org/elasticsearch/xpack/logsdb/LogsDBPlugin.java

@@ -13,6 +13,7 @@ import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.IndexSettingProvider;
+import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.xpack.core.XPackPlugin;
@@ -62,10 +63,16 @@ public class LogsDBPlugin extends Plugin implements ActionPlugin {
         if (DiscoveryNode.isStateless(settings)) {
             return List.of(logsdbIndexModeSettingsProvider);
         }
-        return List.of(
-            new SyntheticSourceIndexSettingsProvider(licenseService, parameters.mapperServiceFactory(), logsdbIndexModeSettingsProvider),
-            logsdbIndexModeSettingsProvider
+        var syntheticSettingProvider = new SyntheticSourceIndexSettingsProvider(
+            licenseService,
+            parameters.mapperServiceFactory(),
+            logsdbIndexModeSettingsProvider,
+            () -> IndexVersion.min(
+                IndexVersion.current(),
+                parameters.clusterService().state().nodes().getMaxDataNodeCompatibleIndexVersion()
+            )
         );
+        return List.of(syntheticSettingProvider, logsdbIndexModeSettingsProvider);
     }
 
     @Override

x-pack/plugin/logsdb/src/main/java/org/elasticsearch/xpack/logsdb/SyntheticSourceIndexSettingsProvider.java

@@ -26,6 +26,7 @@ import org.elasticsearch.index.mapper.SourceFieldMapper;
 import java.io.IOException;
 import java.time.Instant;
 import java.util.List;
+import java.util.function.Supplier;
 
 import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_ROUTING_PATH;
 
@@ -39,15 +40,18 @@ final class SyntheticSourceIndexSettingsProvider implements IndexSettingProvider
     private final SyntheticSourceLicenseService syntheticSourceLicenseService;
     private final CheckedFunction<IndexMetadata, MapperService, IOException> mapperServiceFactory;
     private final LogsdbIndexModeSettingsProvider logsdbIndexModeSettingsProvider;
+    private final Supplier<IndexVersion> createdIndexVersion;
 
     SyntheticSourceIndexSettingsProvider(
         SyntheticSourceLicenseService syntheticSourceLicenseService,
         CheckedFunction<IndexMetadata, MapperService, IOException> mapperServiceFactory,
-        LogsdbIndexModeSettingsProvider logsdbIndexModeSettingsProvider
+        LogsdbIndexModeSettingsProvider logsdbIndexModeSettingsProvider,
+        Supplier<IndexVersion> createdIndexVersion
     ) {
         this.syntheticSourceLicenseService = syntheticSourceLicenseService;
         this.mapperServiceFactory = mapperServiceFactory;
         this.logsdbIndexModeSettingsProvider = logsdbIndexModeSettingsProvider;
+        this.createdIndexVersion = createdIndexVersion;
     }
 
     @Override
@@ -148,7 +152,7 @@ final class SyntheticSourceIndexSettingsProvider implements IndexSettingProvider
         );
         int shardReplicas = indexTemplateAndCreateRequestSettings.getAsInt(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0);
         var finalResolvedSettings = Settings.builder()
-            .put(IndexMetadata.SETTING_VERSION_CREATED, IndexVersion.current())
+            .put(IndexMetadata.SETTING_VERSION_CREATED, createdIndexVersion.get())
             .put(indexTemplateAndCreateRequestSettings)
             .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, dummyShards)
             .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, shardReplicas)

x-pack/plugin/logsdb/src/test/java/org/elasticsearch/xpack/logsdb/SyntheticSourceIndexSettingsProviderTests.java

@@ -15,6 +15,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.MapperTestUtils;
 import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.license.MockLicenseState;
@@ -54,7 +55,7 @@ public class SyntheticSourceIndexSettingsProviderTests extends ESTestCase {
         provider = new SyntheticSourceIndexSettingsProvider(syntheticSourceLicenseService, im -> {
             newMapperServiceCounter.incrementAndGet();
             return MapperTestUtils.newMapperService(xContentRegistry(), createTempDir(), im.getSettings(), im.getIndex().getName());
-        }, getLogsdbIndexModeSettingsProvider(false));
+        }, getLogsdbIndexModeSettingsProvider(false), IndexVersion::current);
         newMapperServiceCounter.set(0);
     }
 
@@ -80,10 +81,12 @@ public class SyntheticSourceIndexSettingsProviderTests extends ESTestCase {
             boolean result = provider.newIndexHasSyntheticSourceUsage(indexName, null, settings, List.of(new CompressedXContent(mapping)));
             assertTrue(result);
             assertThat(newMapperServiceCounter.get(), equalTo(1));
+            assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
         }
         {
             String mapping;
-            if (randomBoolean()) {
+            boolean withSourceMode = randomBoolean();
+            if (withSourceMode) {
                 mapping = """
                     {
                         "_doc": {
@@ -114,6 +117,9 @@ public class SyntheticSourceIndexSettingsProviderTests extends ESTestCase {
             boolean result = provider.newIndexHasSyntheticSourceUsage(indexName, null, settings, List.of(new CompressedXContent(mapping)));
             assertFalse(result);
             assertThat(newMapperServiceCounter.get(), equalTo(2));
+            if (withSourceMode) {
+                assertWarnings(SourceFieldMapper.DEPRECATION_WARNING);
+            }
         }
     }
 
@@ -336,7 +342,8 @@ public class SyntheticSourceIndexSettingsProviderTests extends ESTestCase {
         provider = new SyntheticSourceIndexSettingsProvider(
             syntheticSourceLicenseService,
             im -> MapperTestUtils.newMapperService(xContentRegistry(), createTempDir(), im.getSettings(), im.getIndex().getName()),
-            getLogsdbIndexModeSettingsProvider(true)
+            getLogsdbIndexModeSettingsProvider(true),
+            IndexVersion::current
         );
         final Settings settings = Settings.EMPTY;
 

x-pack/plugin/logsdb/src/yamlRestTest/resources/rest-api-spec/test/40_source_mode_setting.yml

@@ -459,13 +459,7 @@ create an index with time_series index mode and synthetic source:
       indices.get_settings:
         index: "test_time_series_index_mode_synthetic"
   - match: { test_time_series_index_mode_synthetic.settings.index.mode: time_series }
-
-
-  - do:
-      indices.get_mapping:
-        index: test_time_series_index_mode_synthetic
-
-  - match: { test_time_series_index_mode_synthetic.mappings._source.mode: synthetic }
+  - match: { test_time_series_index_mode_synthetic.settings.index.mapping.source.mode: synthetic }
 
 ---
 create an index with logsdb index mode and synthetic source:
@@ -482,12 +476,7 @@ create an index with logsdb index mode and synthetic source:
       indices.get_settings:
         index: "test_logsdb_index_mode_synthetic"
   - match: { test_logsdb_index_mode_synthetic.settings.index.mode: logsdb }
-
-  - do:
-      indices.get_mapping:
-        index: test_logsdb_index_mode_synthetic
-
-  - match: { test_logsdb_index_mode_synthetic.mappings._source.mode: synthetic }
+  - match: { test_logsdb_index_mode_synthetic.settings.index.mapping.source.mode: synthetic }
 
 ---
 create an index with time_series index mode and stored source:
@@ -512,14 +501,9 @@ create an index with time_series index mode and stored source:
   - do:
       indices.get_settings:
         index: "test_time_series_index_mode_undefined"
+  - match: { test_time_series_index_mode_undefined.settings.index.mode: time_series }
   - match: { test_time_series_index_mode_undefined.settings.index.mapping.source.mode: stored }
 
-  - do:
-      indices.get_mapping:
-        index: test_time_series_index_mode_undefined
-
-  - match: { test_time_series_index_mode_undefined.mappings._source.mode: stored }
-
 ---
 create an index with logsdb index mode and stored source:
   - do:
@@ -532,10 +516,10 @@ create an index with logsdb index mode and stored source:
               mapping.source.mode: stored
 
   - do:
-      indices.get_mapping:
-        index: test_logsdb_index_mode_undefined
-
-  - match: { test_logsdb_index_mode_undefined.mappings._source.mode: stored }
+      indices.get_settings:
+        index: "test_logsdb_index_mode_undefined"
+  - match: { test_logsdb_index_mode_undefined.settings.index.mode: logsdb }
+  - match: { test_logsdb_index_mode_undefined.settings.index.mapping.source.mode: stored }
 
 ---
 create an index with time_series index mode and disabled source: