Adding mappings to data streams (#129787)

modules/data-streams/src/main/java/org/elasticsearch/datastreams/action/TransportGetDataStreamsAction.java

@@ -54,6 +54,7 @@ import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.io.IOException;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -261,6 +262,7 @@ public class TransportGetDataStreamsAction extends TransportLocalProjectMetadata
                     Settings settings = dataStream.getEffectiveSettings(state.metadata());
                     ilmPolicyName = settings.get(IndexMetadata.LIFECYCLE_NAME);
                     if (indexMode == null && state.metadata().templatesV2().get(indexTemplate) != null) {
+                        try {
                             indexMode = resolveMode(
                                 state,
                                 indexSettingProviders,
@@ -268,6 +270,9 @@ public class TransportGetDataStreamsAction extends TransportLocalProjectMetadata
                                 settings,
                                 dataStream.getEffectiveIndexTemplate(state.metadata())
                             );
+                        } catch (IOException e) {
+                            throw new RuntimeException(e);
+                        }
                     }
                     indexTemplatePreferIlmValue = PREFER_ILM_SETTING.get(settings);
                 } else {

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

@@ -258,6 +258,7 @@ public class UpdateTimeSeriesRangeServiceTests extends ESTestCase {
                 2,
                 ds2.getMetadata(),
                 ds2.getSettings(),
+                ds2.getMappings(),
                 ds2.isHidden(),
                 ds2.isReplicated(),
                 ds2.isSystem(),

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -323,6 +323,7 @@ public class TransportVersions {
     public static final TransportVersion ML_INFERENCE_ELASTIC_DENSE_TEXT_EMBEDDINGS_ADDED = def(9_109_00_0);
     public static final TransportVersion ML_INFERENCE_COHERE_API_VERSION = def(9_110_0_00);
     public static final TransportVersion ESQL_PROFILE_INCLUDE_PLAN = def(9_111_0_00);
+    public static final TransportVersion MAPPINGS_IN_DATA_STREAMS = def(9_112_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/cluster/metadata/ComposableIndexTemplate.java

@@ -19,18 +19,24 @@ import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.mapper.DataStreamTimestampFieldMapper;
 import org.elasticsearch.index.mapper.MapperService;
+import org.elasticsearch.index.mapper.Mapping;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentFactory;
 import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xcontent.XContentParserConfiguration;
+import org.elasticsearch.xcontent.XContentType;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -51,6 +57,14 @@ public class ComposableIndexTemplate implements SimpleDiffable<ComposableIndexTe
     private static final ParseField ALLOW_AUTO_CREATE = new ParseField("allow_auto_create");
     private static final ParseField IGNORE_MISSING_COMPONENT_TEMPLATES = new ParseField("ignore_missing_component_templates");
     private static final ParseField DEPRECATED = new ParseField("deprecated");
+    public static final CompressedXContent EMPTY_MAPPINGS;
+    static {
+        try {
+            EMPTY_MAPPINGS = new CompressedXContent(Map.of());
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
 
     @SuppressWarnings("unchecked")
     public static final ConstructingObjectParser<ComposableIndexTemplate, Void> PARSER = new ConstructingObjectParser<>(
@@ -338,6 +352,64 @@ public class ComposableIndexTemplate implements SimpleDiffable<ComposableIndexTe
         return mergedIndexTemplateBuilder.build();
     }
 
+    public ComposableIndexTemplate mergeMappings(CompressedXContent mappings) throws IOException {
+        Objects.requireNonNull(mappings);
+        if (Mapping.EMPTY.toCompressedXContent().equals(mappings) && this.template() != null && this.template().mappings() != null) {
+            return this;
+        }
+        ComposableIndexTemplate.Builder mergedIndexTemplateBuilder = this.toBuilder();
+        Template.Builder mergedTemplateBuilder;
+        CompressedXContent templateMappings;
+        if (this.template() == null) {
+            mergedTemplateBuilder = Template.builder();
+            templateMappings = null;
+        } else {
+            mergedTemplateBuilder = Template.builder(this.template());
+            templateMappings = this.template().mappings();
+        }
+        mergedTemplateBuilder.mappings(templateMappings == null ? mappings : merge(templateMappings, mappings));
+        mergedIndexTemplateBuilder.template(mergedTemplateBuilder);
+        return mergedIndexTemplateBuilder.build();
+    }
+
+    @SuppressWarnings("unchecked")
+    private CompressedXContent merge(CompressedXContent originalMapping, CompressedXContent mappingAddition) throws IOException {
+        Map<String, Object> mappingAdditionMap = XContentHelper.convertToMap(mappingAddition.uncompressed(), true, XContentType.JSON).v2();
+        Map<String, Object> combinedMappingMap = new HashMap<>();
+        if (originalMapping != null) {
+            Map<String, Object> originalMappingMap = XContentHelper.convertToMap(originalMapping.uncompressed(), true, XContentType.JSON)
+                .v2();
+            if (originalMappingMap.containsKey(MapperService.SINGLE_MAPPING_NAME)) {
+                combinedMappingMap.putAll((Map<String, ?>) originalMappingMap.get(MapperService.SINGLE_MAPPING_NAME));
+            } else {
+                combinedMappingMap.putAll(originalMappingMap);
+            }
+        }
+        XContentHelper.update(combinedMappingMap, mappingAdditionMap, true);
+        return convertMappingMapToXContent(combinedMappingMap);
+    }
+
+    private static CompressedXContent convertMappingMapToXContent(Map<String, Object> rawAdditionalMapping) throws IOException {
+        CompressedXContent compressedXContent;
+        if (rawAdditionalMapping.isEmpty()) {
+            compressedXContent = EMPTY_MAPPINGS;
+        } else {
+            try (var parser = XContentHelper.mapToXContentParser(XContentParserConfiguration.EMPTY, rawAdditionalMapping)) {
+                compressedXContent = mappingFromXContent(parser);
+            }
+        }
+        return compressedXContent;
+    }
+
+    private static CompressedXContent mappingFromXContent(XContentParser parser) throws IOException {
+        XContentParser.Token token = parser.nextToken();
+        if (token == XContentParser.Token.START_OBJECT) {
+            return new CompressedXContent(Strings.toString(XContentFactory.jsonBuilder().map(parser.mapOrdered())));
+        } else {
+            throw new IllegalArgumentException("Unexpected token: " + token);
+        }
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(

server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java

@@ -29,6 +29,7 @@ import org.elasticsearch.cluster.metadata.DataStreamLifecycle.DownsamplingRound;
 import org.elasticsearch.common.ParsingException;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Settings;
@@ -47,9 +48,11 @@ import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.indices.SystemIndices;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
+import org.elasticsearch.xcontent.ObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentFactory;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
 import org.elasticsearch.xcontent.XContentType;
@@ -58,6 +61,7 @@ import java.io.IOException;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
+import java.util.Base64;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
@@ -70,6 +74,7 @@ import java.util.function.LongSupplier;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
+import static org.elasticsearch.cluster.metadata.ComposableIndexTemplate.EMPTY_MAPPINGS;
 import static org.elasticsearch.cluster.metadata.MetadataCreateDataStreamService.lookupTemplateForDataStream;
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
 import static org.elasticsearch.index.IndexSettings.LIFECYCLE_ORIGINATION_DATE;
@@ -89,6 +94,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
     public static final String FAILURE_STORE_PREFIX = ".fs-";
     public static final DateFormatter DATE_FORMATTER = DateFormatter.forPattern("uuuu.MM.dd");
     public static final String TIMESTAMP_FIELD_NAME = "@timestamp";
+
     // Timeseries indices' leaf readers should be sorted by desc order of their timestamp field, as it allows search time optimizations
     public static final Comparator<LeafReader> TIMESERIES_LEAF_READERS_SORTER = Comparator.comparingLong((LeafReader r) -> {
         try {
@@ -120,6 +126,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
     @Nullable
     private final Map<String, Object> metadata;
     private final Settings settings;
+    private final CompressedXContent mappings;
     private final boolean hidden;
     private final boolean replicated;
     private final boolean system;
@@ -156,6 +163,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
             generation,
             metadata,
             Settings.EMPTY,
+            EMPTY_MAPPINGS,
             hidden,
             replicated,
             system,
@@ -176,6 +184,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
         long generation,
         Map<String, Object> metadata,
         Settings settings,
+        CompressedXContent mappings,
         boolean hidden,
         boolean replicated,
         boolean system,
@@ -192,6 +201,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
             generation,
             metadata,
             settings,
+            mappings,
             hidden,
             replicated,
             system,
@@ -210,6 +220,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
         long generation,
         Map<String, Object> metadata,
         Settings settings,
+        CompressedXContent mappings,
         boolean hidden,
         boolean replicated,
         boolean system,
@@ -225,6 +236,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
         this.generation = generation;
         this.metadata = metadata;
         this.settings = Objects.requireNonNull(settings);
+        this.mappings = Objects.requireNonNull(mappings);
         assert system == false || hidden; // system indices must be hidden
         this.hidden = hidden;
         this.replicated = replicated;
@@ -286,11 +298,18 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
         } else {
             settings = Settings.EMPTY;
         }
+        CompressedXContent mappings;
+        if (in.getTransportVersion().onOrAfter(TransportVersions.MAPPINGS_IN_DATA_STREAMS)) {
+            mappings = CompressedXContent.readCompressedString(in);
+        } else {
+            mappings = EMPTY_MAPPINGS;
+        }
         return new DataStream(
             name,
             generation,
             metadata,
             settings,
+            mappings,
             hidden,
             replicated,
             system,
@@ -381,8 +400,8 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
         return backingIndices.rolloverOnWrite;
     }
 
-    public ComposableIndexTemplate getEffectiveIndexTemplate(ProjectMetadata projectMetadata) {
+    public ComposableIndexTemplate getEffectiveIndexTemplate(ProjectMetadata projectMetadata) throws IOException {
-        return getMatchingIndexTemplate(projectMetadata).mergeSettings(settings);
+        return getMatchingIndexTemplate(projectMetadata).mergeSettings(settings).mergeMappings(mappings);
     }
 
     public Settings getEffectiveSettings(ProjectMetadata projectMetadata) {
@@ -391,6 +410,10 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
         return templateSettings.merge(settings);
     }
 
+    public CompressedXContent getEffectiveMappings(ProjectMetadata projectMetadata) throws IOException {
+        return getMatchingIndexTemplate(projectMetadata).mergeMappings(mappings).template().mappings();
+    }
+
     private ComposableIndexTemplate getMatchingIndexTemplate(ProjectMetadata projectMetadata) {
         return lookupTemplateForDataStream(name, projectMetadata);
     }
@@ -510,6 +533,10 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
         return settings;
     }
 
+    public CompressedXContent getMappings() {
+        return mappings;
+    }
+
     @Override
     public boolean isHidden() {
         return hidden;
@@ -1354,6 +1381,9 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
             || out.getTransportVersion().isPatchFrom(TransportVersions.SETTINGS_IN_DATA_STREAMS_8_19)) {
             settings.writeTo(out);
         }
+        if (out.getTransportVersion().onOrAfter(TransportVersions.MAPPINGS_IN_DATA_STREAMS)) {
+            mappings.writeTo(out);
+        }
     }
 
     public static final ParseField NAME_FIELD = new ParseField("name");
@@ -1376,6 +1406,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
     public static final ParseField FAILURE_AUTO_SHARDING_FIELD = new ParseField("failure_auto_sharding");
     public static final ParseField DATA_STREAM_OPTIONS_FIELD = new ParseField("options");
     public static final ParseField SETTINGS_FIELD = new ParseField("settings");
+    public static final ParseField MAPPINGS_FIELD = new ParseField("mappings");
 
     @SuppressWarnings("unchecked")
     private static final ConstructingObjectParser<DataStream, Void> PARSER = new ConstructingObjectParser<>(
@@ -1385,6 +1416,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
             (Long) args[2],
             (Map<String, Object>) args[3],
             args[17] == null ? Settings.EMPTY : (Settings) args[17],
+            args[18] == null ? EMPTY_MAPPINGS : (CompressedXContent) args[18],
             args[4] != null && (boolean) args[4],
             args[5] != null && (boolean) args[5],
             args[6] != null && (boolean) args[6],
@@ -1456,6 +1488,18 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
             DATA_STREAM_OPTIONS_FIELD
         );
         PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> Settings.fromXContent(p), SETTINGS_FIELD);
+        PARSER.declareField(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> {
+            XContentParser.Token token = p.currentToken();
+            if (token == XContentParser.Token.VALUE_STRING) {
+                return new CompressedXContent(Base64.getDecoder().decode(p.text()));
+            } else if (token == XContentParser.Token.VALUE_EMBEDDED_OBJECT) {
+                return new CompressedXContent(p.binaryValue());
+            } else if (token == XContentParser.Token.START_OBJECT) {
+                return new CompressedXContent(Strings.toString(XContentFactory.jsonBuilder().map(p.mapOrdered())));
+            } else {
+                throw new IllegalArgumentException("Unexpected token: " + token);
+            }
+        }, MAPPINGS_FIELD, ObjectParser.ValueType.VALUE_OBJECT_ARRAY);
     }
 
     public static DataStream fromXContent(XContentParser parser) throws IOException {
@@ -1520,6 +1564,20 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
         builder.startObject(SETTINGS_FIELD.getPreferredName());
         this.settings.toXContent(builder, params);
         builder.endObject();
+
+        String context = params.param(Metadata.CONTEXT_MODE_PARAM, Metadata.CONTEXT_MODE_API);
+        boolean binary = params.paramAsBoolean("binary", false);
+        if (Metadata.CONTEXT_MODE_API.equals(context) || binary == false) {
+            Map<String, Object> uncompressedMapping = XContentHelper.convertToMap(this.mappings.uncompressed(), true, XContentType.JSON)
+                .v2();
+            if (uncompressedMapping.isEmpty() == false) {
+                builder.field(MAPPINGS_FIELD.getPreferredName());
+                builder.map(uncompressedMapping);
+            }
+        } else {
+            builder.field(MAPPINGS_FIELD.getPreferredName(), mappings.compressed());
+        }
+
         builder.endObject();
         return builder;
     }
@@ -1864,6 +1922,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
         @Nullable
         private Map<String, Object> metadata = null;
         private Settings settings = Settings.EMPTY;
+        private CompressedXContent mappings = EMPTY_MAPPINGS;
         private boolean hidden = false;
         private boolean replicated = false;
         private boolean system = false;
@@ -1892,6 +1951,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
             generation = dataStream.generation;
             metadata = dataStream.metadata;
             settings = dataStream.settings;
+            mappings = dataStream.mappings;
             hidden = dataStream.hidden;
             replicated = dataStream.replicated;
             system = dataStream.system;
@@ -1928,6 +1988,11 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
             return this;
         }
 
+        public Builder setMappings(CompressedXContent mappings) {
+            this.mappings = mappings;
+            return this;
+        }
+
         public Builder setHidden(boolean hidden) {
             this.hidden = hidden;
             return this;
@@ -1989,6 +2054,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
                 generation,
                 metadata,
                 settings,
+                mappings,
                 hidden,
                 replicated,
                 system,

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamService.java

@@ -332,6 +332,7 @@ public class MetadataCreateDataStreamService {
             initialGeneration,
             template.metadata() != null ? Map.copyOf(template.metadata()) : null,
             Settings.EMPTY,
+            ComposableIndexTemplate.EMPTY_MAPPINGS,
             hidden,
             false,
             isSystem,

server/src/test/java/org/elasticsearch/cluster/metadata/ComposableIndexTemplateTests.java

@@ -29,6 +29,7 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 
+import static org.elasticsearch.cluster.metadata.ComposableIndexTemplate.EMPTY_MAPPINGS;
 import static org.elasticsearch.cluster.metadata.DataStream.TIMESTAMP_FIELD_NAME;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
@@ -283,9 +284,7 @@ public class ComposableIndexTemplateTests extends SimpleDiffableSerializationTes
     }
 
     public void testMergeEmptySettingsIntoTemplateWithNonEmptySettings() {
-        // We only have settings from the template, so the effective template will just be the original template
+        // Attempting to merge in null settings ought to fail
-        Settings templateSettings = randomSettings();
-        Template.Builder templateBuilder = Template.builder().settings(templateSettings).mappings(randomMappings(null));
         ComposableIndexTemplate indexTemplate = randomInstance();
         expectThrows(NullPointerException.class, () -> indexTemplate.mergeSettings(null));
         assertThat(indexTemplate.mergeSettings(Settings.EMPTY), equalTo(indexTemplate));
@@ -325,12 +324,14 @@ public class ComposableIndexTemplateTests extends SimpleDiffableSerializationTes
             .put("index.setting3", "templateValue")
             .put("index.setting4", "templateValue")
             .build();
+        List<String> componentTemplates = List.of("component_template_1");
         CompressedXContent templateMappings = randomMappings(randomDataStreamTemplate());
         Template.Builder templateBuilder = Template.builder().settings(templateSettings).mappings(templateMappings);
         ComposableIndexTemplate indexTemplate = ComposableIndexTemplate.builder()
             .indexPatterns(List.of(dataStreamName))
             .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
             .template(templateBuilder)
+            .componentTemplates(componentTemplates)
             .build();
         Settings mergedSettings = Settings.builder()
             .put("index.setting1", "dataStreamValue")
@@ -342,7 +343,62 @@ public class ComposableIndexTemplateTests extends SimpleDiffableSerializationTes
             .indexPatterns(List.of(dataStreamName))
             .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
             .template(expectedTemplateBuilder)
+            .componentTemplates(componentTemplates)
             .build();
         assertThat(indexTemplate.mergeSettings(dataStreamSettings), equalTo(expectedEffectiveTemplate));
     }
+
+    public void testMergeEmptyMappingsIntoTemplateWithNonEmptySettings() throws IOException {
+        // Attempting to merge in null mappings ought to fail
+        ComposableIndexTemplate indexTemplate = randomInstance();
+        expectThrows(NullPointerException.class, () -> indexTemplate.mergeMappings(null));
+        assertThat(indexTemplate.mergeMappings(EMPTY_MAPPINGS), equalTo(indexTemplate));
+        assertThat(indexTemplate.mergeSettings(Settings.EMPTY), equalTo(indexTemplate));
+    }
+
+    public void testMergeNonEmptyMappingsIntoTemplateWithEmptyMappings() throws IOException {
+        // We only have settings from the data stream, so we expect to get only those back in the effective template
+        CompressedXContent dataStreamMappings = randomMappings(randomDataStreamTemplate());
+        String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        Settings templateSettings = Settings.EMPTY;
+        CompressedXContent templateMappings = new CompressedXContent(Map.of("_doc", Map.of()));
+        Template.Builder templateBuilder = Template.builder().settings(templateSettings).mappings(templateMappings);
+        ComposableIndexTemplate indexTemplate = ComposableIndexTemplate.builder()
+            .indexPatterns(List.of(dataStreamName))
+            .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
+            .template(templateBuilder)
+            .build();
+        Template.Builder expectedTemplateBuilder = Template.builder().settings(templateSettings).mappings(dataStreamMappings);
+        ComposableIndexTemplate expectedEffectiveTemplate = ComposableIndexTemplate.builder()
+            .indexPatterns(List.of(dataStreamName))
+            .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
+            .template(expectedTemplateBuilder)
+            .build();
+        assertThat(indexTemplate.mergeMappings(dataStreamMappings), equalTo(expectedEffectiveTemplate));
+    }
+
+    public void testMergeMappings() throws IOException {
+        // Here we have settings from both the template and the data stream, so we expect the data stream settings to take precedence
+        CompressedXContent dataStreamMappings = new CompressedXContent(Map.of());
+        String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        CompressedXContent templateMappings = new CompressedXContent(Map.of("_doc", Map.of()));
+        Settings templateSettings = randomSettings();
+        List<String> componentTemplates = List.of("component_template_1");
+        Template.Builder templateBuilder = Template.builder().settings(templateSettings).mappings(templateMappings);
+        ComposableIndexTemplate indexTemplate = ComposableIndexTemplate.builder()
+            .indexPatterns(List.of(dataStreamName))
+            .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
+            .template(templateBuilder)
+            .componentTemplates(componentTemplates)
+            .build();
+        Template.Builder expectedTemplateBuilder = Template.builder().settings(templateSettings).mappings(EMPTY_MAPPINGS);
+        ComposableIndexTemplate expectedEffectiveTemplate = ComposableIndexTemplate.builder()
+            .indexPatterns(List.of(dataStreamName))
+            .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
+            .template(expectedTemplateBuilder)
+            .componentTemplates(componentTemplates)
+            .build();
+        ComposableIndexTemplate merged = indexTemplate.mergeMappings(dataStreamMappings);
+        assertThat(merged, equalTo(expectedEffectiveTemplate));
+    }
 }

server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamTests.java

@@ -53,7 +53,6 @@ import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
-import static org.elasticsearch.cluster.metadata.ComponentTemplateTests.randomMappings;
 import static org.elasticsearch.cluster.metadata.DataStream.getDefaultBackingIndexName;
 import static org.elasticsearch.cluster.metadata.DataStream.getDefaultFailureStoreName;
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.newInstance;
@@ -98,6 +97,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
         var generation = instance.getGeneration();
         var metadata = instance.getMetadata();
         var settings = instance.getSettings();
+        var mappings = instance.getMappings();
         var isHidden = instance.isHidden();
         var isReplicated = instance.isReplicated();
         var isSystem = instance.isSystem();
@@ -110,7 +110,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
         var autoShardingEvent = instance.getAutoShardingEvent();
         var failureRolloverOnWrite = instance.getFailureComponent().isRolloverOnWrite();
         var failureAutoShardingEvent = instance.getDataComponent().getAutoShardingEvent();
-        switch (between(0, 16)) {
+        switch (between(0, 17)) {
             case 0 -> name = randomAlphaOfLength(10);
             case 1 -> indices = randomNonEmptyIndexInstances();
             case 2 -> generation = instance.getGeneration() + randomIntBetween(1, 10);
@@ -179,6 +179,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
                 ? null
                 : new DataStreamAutoShardingEvent(indices.getLast().getName(), randomIntBetween(1, 10), randomMillisUpToYear9999());
             case 16 -> settings = randomValueOtherThan(settings, DataStreamTestHelper::randomSettings);
+            case 17 -> mappings = randomValueOtherThan(mappings, ComponentTemplateTests::randomMappings);
         }
 
         return new DataStream(
@@ -186,6 +187,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             generation,
             metadata,
             settings,
+            mappings,
             isHidden,
             isReplicated,
             isSystem,
@@ -1948,6 +1950,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             generation,
             metadata,
             randomSettings(),
+            randomMappings(),
             isSystem,
             randomBoolean(),
             isSystem,
@@ -2141,6 +2144,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             randomNonNegativeInt(),
             null,
             randomSettings(),
+            randomMappings(),
             hidden,
             replicated,
             system,
@@ -2160,6 +2164,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             randomNonNegativeInt(),
             null,
             randomSettings(),
+            randomMappings(),
             hidden,
             replicated,
             system,
@@ -2186,6 +2191,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             randomNonNegativeInt(),
             null,
             randomSettings(),
+            randomMappings(),
             hidden,
             replicated,
             system,
@@ -2211,6 +2217,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             randomNonNegativeInt(),
             null,
             randomSettings(),
+            randomMappings(),
             hidden,
             replicated,
             system,
@@ -2234,6 +2241,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             randomNonNegativeInt(),
             null,
             randomSettings(),
+            randomMappings(),
             hidden,
             replicated,
             system,
@@ -2266,6 +2274,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             randomNonNegativeInt(),
             null,
             randomSettings(),
+            randomMappings(),
             hidden,
             replicated,
             system,
@@ -2529,13 +2538,31 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             .indexPatterns(List.of(dataStream.getName()))
             .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
             .template(templateBuilder)
+            .componentTemplates(List.of("component-template-1"))
             .build();
         ProjectMetadata.Builder projectMetadataBuilder = ProjectMetadata.builder(randomProjectIdOrDefault())
-            .indexTemplates(Map.of(dataStream.getName(), indexTemplate));
+            .indexTemplates(Map.of(dataStream.getName(), indexTemplate))
+            .componentTemplates(
+                Map.of(
+                    "component-template-1",
+                    new ComponentTemplate(
+                        Template.builder()
+                            .settings(
+                                Settings.builder()
+                                    .put("index.setting1", "componentTemplateValue")
+                                    .put("index.setting5", "componentTemplateValue")
+                            )
+                            .build(),
+                        1L,
+                        Map.of()
+                    )
+                )
+            );
         Settings mergedSettings = Settings.builder()
             .put("index.setting1", "dataStreamValue")
             .put("index.setting2", "dataStreamValue")
             .put("index.setting4", "templateValue")
+            .put("index.setting5", "componentTemplateValue")
             .build();
         assertThat(dataStream.getEffectiveSettings(projectMetadataBuilder.build()), equalTo(mergedSettings));
     }
@@ -2547,28 +2574,40 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
         assertThrows(IllegalArgumentException.class, () -> dataStream.getEffectiveIndexTemplate(projectMetadataBuilder.build()));
     }
 
-    public void testGetEffectiveIndexTemplateTemplateSettingsOnly() {
+    public void testGetEffectiveIndexTemplateTemplateNoOverrides() throws IOException {
-        // We only have settings from the template, so the effective template will just be the original template
+        // We only have settings and mappings from the template, so the effective template will just be the original template
-        DataStream dataStream = createDataStream(Settings.EMPTY);
+        DataStream dataStream = createDataStream(Settings.EMPTY, ComposableIndexTemplate.EMPTY_MAPPINGS);
         Settings templateSettings = randomSettings();
         Template.Builder templateBuilder = Template.builder().settings(templateSettings).mappings(randomMappings());
         ComposableIndexTemplate indexTemplate = ComposableIndexTemplate.builder()
             .indexPatterns(List.of(dataStream.getName()))
             .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
             .template(templateBuilder)
+            .componentTemplates(List.of("component-template-1"))
             .build();
         ProjectMetadata.Builder projectMetadataBuilder = ProjectMetadata.builder(randomProjectIdOrDefault())
-            .indexTemplates(Map.of(dataStream.getName(), indexTemplate));
+            .indexTemplates(Map.of(dataStream.getName(), indexTemplate))
+            .componentTemplates(
+                Map.of(
+                    "component-template-1",
+                    new ComponentTemplate(
+                        Template.builder().settings(Settings.builder().put("index.setting5", "componentTemplateValue")).build(),
+                        1L,
+                        Map.of()
+                    )
+                )
+            );
         assertThat(dataStream.getEffectiveIndexTemplate(projectMetadataBuilder.build()), equalTo(indexTemplate));
     }
 
-    public void testGetEffectiveIndexTemplateDataStreamSettingsOnly() {
+    public void testGetEffectiveIndexTemplateDataStreamSettingsOnly() throws IOException {
         // We only have settings from the data stream, so we expect to get only those back in the effective template
         Settings dataStreamSettings = randomSettings();
-        DataStream dataStream = createDataStream(dataStreamSettings);
+        DataStream dataStream = createDataStream(dataStreamSettings, ComposableIndexTemplate.EMPTY_MAPPINGS);
         Settings templateSettings = Settings.EMPTY;
         CompressedXContent templateMappings = randomMappings();
         Template.Builder templateBuilder = Template.builder().settings(templateSettings).mappings(templateMappings);
+
         ComposableIndexTemplate indexTemplate = ComposableIndexTemplate.builder()
             .indexPatterns(List.of(dataStream.getName()))
             .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
@@ -2585,20 +2624,80 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
         assertThat(dataStream.getEffectiveIndexTemplate(projectMetadataBuilder.build()), equalTo(expectedEffectiveTemplate));
     }
 
-    public void testGetEffectiveIndexTemplate() {
+    public void testGetEffectiveIndexTemplate() throws IOException {
         // Here we have settings from both the template and the data stream, so we expect the data stream settings to take precedence
         Settings dataStreamSettings = Settings.builder()
             .put("index.setting1", "dataStreamValue")
             .put("index.setting2", "dataStreamValue")
             .put("index.setting3", (String) null) // This one gets removed from the effective settings
             .build();
-        DataStream dataStream = createDataStream(dataStreamSettings);
+        CompressedXContent dataStreamMappings = new CompressedXContent(
+            Map.of("properties", Map.of("field2", Map.of("type", "text"), "field3", Map.of("type", "keyword")))
+        );
+        DataStream dataStream = createDataStream(dataStreamSettings, dataStreamMappings);
         Settings templateSettings = Settings.builder()
             .put("index.setting1", "templateValue")
             .put("index.setting3", "templateValue")
             .put("index.setting4", "templateValue")
             .build();
-        CompressedXContent templateMappings = randomMappings();
+        CompressedXContent templateMappings = new CompressedXContent(
+            Map.of("_doc", Map.of("properties", Map.of("field1", Map.of("type", "keyword"), "field2", Map.of("type", "keyword"))))
+        );
+        Template.Builder templateBuilder = Template.builder().settings(templateSettings).mappings(templateMappings);
+        List<String> componentTemplates = List.of("component-template-1");
+        ComposableIndexTemplate indexTemplate = ComposableIndexTemplate.builder()
+            .indexPatterns(List.of(dataStream.getName()))
+            .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
+            .template(templateBuilder)
+            .componentTemplates(componentTemplates)
+            .build();
+        ProjectMetadata.Builder projectMetadataBuilder = ProjectMetadata.builder(randomProjectIdOrDefault())
+            .indexTemplates(Map.of(dataStream.getName(), indexTemplate))
+            .componentTemplates(
+                Map.of(
+                    "component-template-1",
+                    new ComponentTemplate(
+                        Template.builder().settings(Settings.builder().put("index.setting5", "componentTemplateValue")).build(),
+                        1L,
+                        Map.of()
+                    )
+                )
+            );
+        Settings mergedSettings = Settings.builder()
+            .put("index.setting1", "dataStreamValue")
+            .put("index.setting2", "dataStreamValue")
+            .put("index.setting4", "templateValue")
+            .build();
+        CompressedXContent mergedMappings = new CompressedXContent(
+            Map.of(
+                "properties",
+                Map.of("field1", Map.of("type", "keyword"), "field2", Map.of("type", "text"), "field3", Map.of("type", "keyword"))
+            )
+        );
+        Template.Builder expectedTemplateBuilder = Template.builder().settings(mergedSettings).mappings(mergedMappings);
+        ComposableIndexTemplate expectedEffectiveTemplate = ComposableIndexTemplate.builder()
+            .indexPatterns(List.of(dataStream.getName()))
+            .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
+            .template(expectedTemplateBuilder)
+            .componentTemplates(componentTemplates)
+            .build();
+        assertThat(dataStream.getEffectiveIndexTemplate(projectMetadataBuilder.build()), equalTo(expectedEffectiveTemplate));
+    }
+
+    public void testGetEffectiveMappingsNoMatchingTemplate() {
+        // No matching template, so we expect an IllegalArgumentException
+        DataStream dataStream = createTestInstance();
+        ProjectMetadata.Builder projectMetadataBuilder = ProjectMetadata.builder(randomProjectIdOrDefault());
+        assertThrows(IllegalArgumentException.class, () -> dataStream.getEffectiveMappings(projectMetadataBuilder.build()));
+    }
+
+    public void testGetEffectiveIndexTemplateDataStreamMappingsOnly() throws IOException {
+        // We only have mappings from the data stream, so we expect to get only those back in the effective template
+        CompressedXContent dataStreamMappings = randomMappings();
+        DataStream dataStream = createDataStream(Settings.EMPTY, dataStreamMappings);
+        Settings templateSettings = Settings.EMPTY;
+        CompressedXContent templateMappings = new CompressedXContent(Map.of("_doc", Map.of()));
+        ;
         Template.Builder templateBuilder = Template.builder().settings(templateSettings).mappings(templateMappings);
         ComposableIndexTemplate indexTemplate = ComposableIndexTemplate.builder()
             .indexPatterns(List.of(dataStream.getName()))
@@ -2607,12 +2706,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             .build();
         ProjectMetadata.Builder projectMetadataBuilder = ProjectMetadata.builder(randomProjectIdOrDefault())
             .indexTemplates(Map.of(dataStream.getName(), indexTemplate));
-        Settings mergedSettings = Settings.builder()
+        Template.Builder expectedTemplateBuilder = Template.builder().settings(templateSettings).mappings(dataStreamMappings);
-            .put("index.setting1", "dataStreamValue")
-            .put("index.setting2", "dataStreamValue")
-            .put("index.setting4", "templateValue")
-            .build();
-        Template.Builder expectedTemplateBuilder = Template.builder().settings(mergedSettings).mappings(templateMappings);
         ComposableIndexTemplate expectedEffectiveTemplate = ComposableIndexTemplate.builder()
             .indexPatterns(List.of(dataStream.getName()))
             .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
@@ -2621,11 +2715,30 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
         assertThat(dataStream.getEffectiveIndexTemplate(projectMetadataBuilder.build()), equalTo(expectedEffectiveTemplate));
     }
 
+    private static CompressedXContent randomMappings() {
+        try {
+            return new CompressedXContent("{\"_doc\": {\"properties\":{\"" + randomAlphaOfLength(5) + "\":{\"type\":\"keyword\"}}}}");
+        } catch (IOException e) {
+            fail("got an IO exception creating fake mappings: " + e);
+            return null;
+        }
+    }
+
     private DataStream createDataStream(Settings settings) {
         DataStream dataStream = createTestInstance();
         return dataStream.copy().setSettings(settings).build();
     }
 
+    private DataStream createDataStream(CompressedXContent mappings) {
+        DataStream dataStream = createTestInstance();
+        return dataStream.copy().setMappings(mappings).build();
+    }
+
+    private DataStream createDataStream(Settings settings, CompressedXContent mappings) {
+        DataStream dataStream = createTestInstance();
+        return dataStream.copy().setSettings(settings).setMappings(mappings).build();
+    }
+
     private record DataStreamMetadata(Long creationTimeInMillis, Long rolloverTimeInMillis, Long originationTimeInMillis) {
         public static DataStreamMetadata dataStreamMetadata(Long creationTimeInMillis, Long rolloverTimeInMillis) {
             return new DataStreamMetadata(creationTimeInMillis, rolloverTimeInMillis, null);
@@ -2669,4 +2782,19 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             builder.put(im, false);
         }
     }
+
+    @Override
+    protected ToXContent.Params getToXContentParams() {
+        if (randomBoolean()) {
+            return ToXContent.EMPTY_PARAMS;
+        }
+        return new ToXContent.MapParams(
+            Map.of(
+                "binary",
+                randomFrom("true", "false"),
+                Metadata.CONTEXT_MODE_PARAM,
+                randomFrom(Metadata.XContentContext.values()).toString()
+            )
+        );
+    }
 }

test/framework/src/main/java/org/elasticsearch/cluster/metadata/DataStreamTestHelper.java

@@ -17,6 +17,7 @@ import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.routing.allocation.WriteLoadForecaster;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.settings.IndexScopedSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.CheckedFunction;
@@ -58,6 +59,7 @@ import org.hamcrest.Description;
 import org.hamcrest.Matcher;
 import org.hamcrest.TypeSafeMatcher;
 
+import java.io.IOException;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -85,6 +87,7 @@ import static org.elasticsearch.test.ESTestCase.randomIntBetween;
 import static org.elasticsearch.test.ESTestCase.randomMap;
 import static org.elasticsearch.test.ESTestCase.randomMillisUpToYear9999;
 import static org.elasticsearch.test.ESTestCase.randomPositiveTimeValue;
+import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -365,6 +368,7 @@ public final class DataStreamTestHelper {
             generation,
             metadata,
             randomSettings(),
+            randomMappings(),
             system ? true : randomBoolean(),
             replicated,
             system,
@@ -400,6 +404,15 @@ public final class DataStreamTestHelper {
         );
     }
 
+    private static CompressedXContent randomMappings() {
+        try {
+            return new CompressedXContent("{\"properties\":{\"" + randomAlphaOfLength(5) + "\":{\"type\":\"keyword\"}}}");
+        } catch (IOException e) {
+            fail("got an IO exception creating fake mappings: " + e);
+            return null;
+        }
+    }
+
     public static DataStreamAlias randomAliasInstance() {
         List<String> dataStreams = List.of(generateRandomStringArray(5, 5, false, false));
         return new DataStreamAlias(