Computing full effective mappings for data stream mapping APIs

Author: masseyke

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

@@ -22,10 +22,12 @@
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xcontent.NamedXContentRegistry;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -35,6 +37,8 @@ public class TransportGetDataStreamMappingsAction extends TransportLocalProjectM
     GetDataStreamMappingsAction.Request,
     GetDataStreamMappingsAction.Response> {
     private final IndexNameExpressionResolver indexNameExpressionResolver;
+    private final NamedXContentRegistry xContentRegistry;
+    private final IndicesService indicesService;
 
     @Inject
     public TransportGetDataStreamMappingsAction(
@@ -43,7 +47,9 @@ public TransportGetDataStreamMappingsAction(
         ThreadPool threadPool,
         ActionFilters actionFilters,
         ProjectResolver projectResolver,
-        IndexNameExpressionResolver indexNameExpressionResolver
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        NamedXContentRegistry xContentRegistry,
+        IndicesService indicesService
     ) {
         super(
             GetSettingsAction.NAME,
@@ -54,6 +60,8 @@ public TransportGetDataStreamMappingsAction(
             projectResolver
         );
         this.indexNameExpressionResolver = indexNameExpressionResolver;
+        this.xContentRegistry = xContentRegistry;
+        this.indicesService = indicesService;
     }
 
     @Override
@@ -81,7 +89,7 @@ protected void localClusterStateOperation(
                 new GetDataStreamMappingsAction.DataStreamMappingsResponse(
                     dataStreamName,
                     dataStream.getMappings(),
-                    dataStream.getEffectiveMappings(project.metadata())
+                    dataStream.getEffectiveMappings(project.metadata(), xContentRegistry, indicesService)
                 )
             );
         }

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

@@ -29,13 +29,14 @@
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.mapper.Mapping;
+import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.indices.SystemIndices;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xcontent.NamedXContentRegistry;
 
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -47,6 +48,8 @@ public class TransportUpdateDataStreamMappingsAction extends TransportMasterNode
     private final IndexNameExpressionResolver indexNameExpressionResolver;
     private final SystemIndices systemIndices;
     private final ProjectResolver projectResolver;
+    private final NamedXContentRegistry xContentRegistry;
+    private final IndicesService indicesService;
 
     @Inject
     public TransportUpdateDataStreamMappingsAction(
@@ -57,7 +60,9 @@ public TransportUpdateDataStreamMappingsAction(
         ProjectResolver projectResolver,
         MetadataDataStreamsService metadataDataStreamsService,
         IndexNameExpressionResolver indexNameExpressionResolver,
-        SystemIndices systemIndices
+        SystemIndices systemIndices,
+        NamedXContentRegistry xContentRegistry,
+        IndicesService indicesService
     ) {
         super(
             UpdateDataStreamMappingsAction.NAME,
@@ -73,6 +78,8 @@ public TransportUpdateDataStreamMappingsAction(
         this.metadataDataStreamsService = metadataDataStreamsService;
         this.indexNameExpressionResolver = indexNameExpressionResolver;
         this.systemIndices = systemIndices;
+        this.xContentRegistry = xContentRegistry;
+        this.indicesService = indicesService;
     }
 
     @Override
@@ -163,10 +170,14 @@ private void updateSingleDataStream(
                                 true,
                                 null,
                                 mappingsOverrides,
-                                dataStream.getEffectiveMappings(clusterService.state().metadata().getProject(projectId))
+                                dataStream.getEffectiveMappings(
+                                    clusterService.state().metadata().getProject(projectId),
+                                    xContentRegistry,
+                                    indicesService
+                                )
                             )
                         );
-                    } catch (IOException e) {
+                    } catch (Exception e) {
                         dataStreamMappingsResponseActionListener.onResponse(
                             new UpdateDataStreamMappingsAction.DataStreamMappingsResponse(
                                 dataStreamName,

modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/250_data_stream_mappings.yml

@@ -37,7 +37,7 @@ setup:
         name: my-data-stream-1
   - match: { data_streams.0.name: my-data-stream-1 }
   - match: { data_streams.0.mappings: {} }
-  - length: { data_streams.0.effective_mappings.properties: 1 }
+  - length: { data_streams.0.effective_mappings.properties: 2 }
 
   - do:
       indices.get_data_stream:
@@ -74,7 +74,7 @@ setup:
       indices.get_data_stream_mappings:
         name: my-data-stream-1
   - match: { data_streams.0.name: my-data-stream-1 }
-  - length: { data_streams.0.effective_mappings.properties: 2 }
+  - length: { data_streams.0.effective_mappings.properties: 3 }
   - match: { data_streams.0.mappings.properties.name.type: "keyword" }
   - match: { data_streams.0.effective_mappings.properties.name.type: "keyword" }
 
@@ -92,3 +92,113 @@ setup:
         index: my-data-stream-1
   - match: { .$oldIndexName.mappings.properties.name: null }
   - match: { .$newIndexName.mappings.properties.name.type: "keyword" }
+
+---
+"Test mappings component templates only":
+  - requires:
+      cluster_features: [ "logs_stream" ]
+      reason: requires setting 'logs_stream' to get or set data stream settings
+
+  - do:
+      cluster.put_component_template:
+        name: mappings-template
+        body:
+          template:
+            mappings:
+              properties:
+                field1:
+                  type: keyword
+                field2:
+                  type: keyword
+
+  - do:
+      allowed_warnings:
+        - "index template [my-component-only-template] has index patterns [my-component-only-data-stream-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-component-only-template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: my-component-only-template
+        body:
+          index_patterns: [ my-component-only-data-stream-* ]
+          data_stream: { }
+          composed_of:
+            - mappings-template
+
+  - do:
+      indices.create_data_stream:
+        name: my-component-only-data-stream-1
+
+  - do:
+      cluster.health:
+        index: "my-component-only-data-stream-1"
+        wait_for_status: green
+
+  - do:
+      indices.get_data_stream:
+        name: my-component-only-data-stream-1
+  - match: { data_streams.0.name: my-component-only-data-stream-1 }
+  - match: { data_streams.0.mappings: {} }
+  - match: { data_streams.0.effective_mappings: null }
+
+  - do:
+      indices.put_data_stream_mappings:
+        name: my-component-only-data-stream-1
+        body:
+          properties:
+            field1:
+              type: text
+            field3:
+              type: text
+  - match: { data_streams.0.name: my-component-only-data-stream-1 }
+  - match: { data_streams.0.applied_to_data_stream: true }
+  - match: { data_streams.0.mappings.properties.field1.type: "text" }
+  - match: { data_streams.0.mappings.properties.field2: null }
+  - match: { data_streams.0.mappings.properties.field3.type: "text" }
+  - match: { data_streams.0.effective_mappings.properties.field1.type: "text" }
+  - match: { data_streams.0.effective_mappings.properties.field2.type: "keyword" }
+  - match: { data_streams.0.effective_mappings.properties.field3.type: "text" }
+
+  - do:
+      indices.rollover:
+        alias: "my-component-only-data-stream-1"
+
+  - do:
+      cluster.health:
+        index: "my-component-only-data-stream-1"
+        wait_for_status: green
+
+  - do:
+      indices.get_data_stream_mappings:
+        name: my-component-only-data-stream-1
+  - match: { data_streams.0.name: my-component-only-data-stream-1 }
+  - length: { data_streams.0.effective_mappings.properties: 4 }
+  - match: { data_streams.0.mappings.properties.field1.type: "text" }
+  - match: { data_streams.0.effective_mappings.properties.field3.type: "text" }
+
+  - do:
+      indices.get_data_stream:
+        name: my-component-only-data-stream-1
+  - match: { data_streams.0.name: my-component-only-data-stream-1 }
+  - match: { data_streams.0.mappings.properties.field1.type: "text" }
+  - match: { data_streams.0.effective_mappings: null }
+  - set: { data_streams.0.indices.0.index_name: oldIndexName }
+  - set: { data_streams.0.indices.1.index_name: newIndexName }
+
+  - do:
+      indices.get_mapping:
+        index: my-component-only-data-stream-1
+  - match: { .$oldIndexName.mappings.properties.field1.type: "keyword" }
+  - match: { .$oldIndexName.mappings.properties.field2.type: "keyword" }
+  - match: { .$oldIndexName.mappings.properties.field3: null }
+  - match: { .$newIndexName.mappings.properties.field1.type: "text" }
+  - match: { .$newIndexName.mappings.properties.field2.type: "keyword" }
+  - match: { .$newIndexName.mappings.properties.field3.type: "text" }
+
+  - do:
+      indices.put_data_stream_mappings:
+        name: my-component-only-data-stream-1
+        body: {}
+  - match: { data_streams.0.name: my-component-only-data-stream-1 }
+  - match: { data_streams.0.applied_to_data_stream: true }
+  - match: { data_streams.0.mappings null }
+  - match: { data_streams.0.effective_mappings.properties.field1.type: "keyword" }
+  - match: { data_streams.0.effective_mappings.properties.field2.type: "keyword" }
+  - match: { data_streams.0.effective_mappings.properties.field3: null }

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

@@ -373,7 +373,7 @@ public ComposableIndexTemplate mergeMappings(CompressedXContent mappings) throws
     }
 
     @SuppressWarnings("unchecked")
-    private CompressedXContent merge(CompressedXContent originalMapping, CompressedXContent mappingAddition) throws IOException {
+    public static 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) {
@@ -389,7 +389,7 @@ private CompressedXContent merge(CompressedXContent originalMapping, CompressedX
         return convertMappingMapToXContent(combinedMappingMap);
     }
 
-    private static CompressedXContent convertMappingMapToXContent(Map<String, Object> rawAdditionalMapping) throws IOException {
+    public static CompressedXContent convertMappingMapToXContent(Map<String, ?> rawAdditionalMapping) throws IOException {
         CompressedXContent compressedXContent;
         if (rawAdditionalMapping.isEmpty()) {
             compressedXContent = EMPTY_MAPPINGS;

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

@@ -46,8 +46,12 @@
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.mapper.DateFieldMapper;
+import org.elasticsearch.index.mapper.DocumentMapper;
+import org.elasticsearch.index.mapper.MapperService;
+import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.indices.SystemIndices;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
+import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.ObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContentObject;
@@ -73,7 +77,9 @@
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.cluster.metadata.ComposableIndexTemplate.EMPTY_MAPPINGS;
+import static org.elasticsearch.cluster.metadata.ComposableIndexTemplate.convertMappingMapToXContent;
 import static org.elasticsearch.cluster.metadata.MetadataCreateDataStreamService.lookupTemplateForDataStream;
+import static org.elasticsearch.cluster.metadata.MetadataCreateIndexService.collectV2Mappings;
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
 import static org.elasticsearch.index.IndexSettings.LIFECYCLE_ORIGINATION_DATE;
 import static org.elasticsearch.index.IndexSettings.PREFER_ILM_SETTING;
@@ -408,8 +414,50 @@ public Settings getEffectiveSettings(ProjectMetadata projectMetadata) {
         return templateSettings.merge(settings);
     }
 
-    public CompressedXContent getEffectiveMappings(ProjectMetadata projectMetadata) throws IOException {
-        return getMatchingIndexTemplate(projectMetadata).mergeMappings(mappings).template().mappings();
+    public CompressedXContent getEffectiveMappings(
+        ProjectMetadata projectMetadata,
+        NamedXContentRegistry xContentRegistry,
+        IndicesService indicesService
+    ) throws Exception {
+        ComposableIndexTemplate templateWithMergedMappings = getEffectiveIndexTemplate(projectMetadata);
+        return getEffectiveMappings(projectMetadata, templateWithMergedMappings, getWriteIndex(), xContentRegistry, indicesService);
+    }
+
+    @SuppressWarnings("unchecked")
+    public static CompressedXContent getEffectiveMappings(
+        ProjectMetadata projectMetadata,
+        ComposableIndexTemplate templateWithMergedMappings,
+        Index writeIndex,
+        NamedXContentRegistry xContentRegistry,
+        IndicesService indicesService
+    ) throws Exception {
+        final List<CompressedXContent> mappings = collectV2Mappings(
+            "{}",
+            projectMetadata,
+            templateWithMergedMappings,
+            xContentRegistry,
+            writeIndex.getName()
+        );
+        return indicesService.withTempIndexService(projectMetadata.index(writeIndex), indexService -> {
+            MapperService mapperService = indexService.mapperService();
+            DocumentMapper documentMapper = mapperService.merge(
+                MapperService.SINGLE_MAPPING_NAME,
+                mappings,
+                MapperService.MergeReason.INDEX_TEMPLATE
+            );
+            Map<String, ?> resultMapping;
+            Map<String, Object> originalMappingMap = XContentHelper.convertToMap(
+                documentMapper.mappingSource().uncompressed(),
+                true,
+                XContentType.JSON
+            ).v2();
+            if (originalMappingMap.containsKey(MapperService.SINGLE_MAPPING_NAME)) {
+                resultMapping = (Map<String, ?>) originalMappingMap.get(MapperService.SINGLE_MAPPING_NAME);
+            } else {
+                resultMapping = originalMappingMap;
+            }
+            return convertMappingMapToXContent(resultMapping);
+        });
     }
 
     private ComposableIndexTemplate getMatchingIndexTemplate(ProjectMetadata projectMetadata) {

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

@@ -38,6 +38,7 @@
 import org.elasticsearch.logging.Logger;
 import org.elasticsearch.snapshots.SnapshotInProgressException;
 import org.elasticsearch.snapshots.SnapshotsService;
+import org.elasticsearch.xcontent.NamedXContentRegistry;
 
 import java.io.IOException;
 import java.util.HashSet;
@@ -56,6 +57,7 @@ public class MetadataDataStreamsService {
     private static final Logger LOGGER = LogManager.getLogger(MetadataDataStreamsService.class);
     private final ClusterService clusterService;
     private final IndicesService indicesService;
+    private final NamedXContentRegistry xContentRegistry;
     private final DataStreamGlobalRetentionSettings globalRetentionSettings;
     private final MasterServiceTaskQueue<UpdateLifecycleTask> updateLifecycleTaskQueue;
     private final MasterServiceTaskQueue<SetRolloverOnWriteTask> setRolloverOnWriteTaskQueue;
@@ -66,10 +68,12 @@ public class MetadataDataStreamsService {
     public MetadataDataStreamsService(
         ClusterService clusterService,
         IndicesService indicesService,
+        NamedXContentRegistry xContentRegistry,
         DataStreamGlobalRetentionSettings globalRetentionSettings
     ) {
         this.clusterService = clusterService;
         this.indicesService = indicesService;
+        this.xContentRegistry = xContentRegistry;
         this.globalRetentionSettings = globalRetentionSettings;
         ClusterStateTaskExecutor<UpdateLifecycleTask> updateLifecycleExecutor = new SimpleBatchedAckListenerTaskExecutor<>() {
 
@@ -497,11 +501,7 @@ private DataStream createDataStreamForUpdatedDataStreamSettings(
         final ComposableIndexTemplate template = lookupTemplateForDataStream(dataStreamName, projectMetadata);
         Settings templateSettings = MetadataIndexTemplateService.resolveSettings(template, projectMetadata.componentTemplates());
         Settings mergedEffectiveSettings = templateSettings.merge(mergedDataStreamSettings);
-        MetadataIndexTemplateService.validateTemplate(
-            mergedEffectiveSettings,
-            dataStream.getEffectiveMappings(projectMetadata),
-            indicesService
-        );
+        MetadataIndexTemplateService.validateTemplate(mergedEffectiveSettings, null, indicesService);
 
         return dataStream.copy().setSettings(mergedDataStreamSettings).build();
     }
@@ -518,11 +518,14 @@ private DataStream createDataStreamForUpdatedDataStreamMappings(
 
         final ComposableIndexTemplate template = lookupTemplateForDataStream(dataStreamName, projectMetadata);
         ComposableIndexTemplate mergedTemplate = template.mergeMappings(mappingsOverrides);
-        MetadataIndexTemplateService.validateTemplate(
-            dataStream.getEffectiveSettings(projectMetadata),
-            mergedTemplate.template().mappings(),
+        CompressedXContent effectiveMappings = DataStream.getEffectiveMappings(
+            projectMetadata,
+            mergedTemplate,
+            dataStream.getWriteIndex(),
+            xContentRegistry,
             indicesService
         );
+        MetadataIndexTemplateService.validateTemplate(dataStream.getEffectiveSettings(projectMetadata), effectiveMappings, indicesService);
         return dataStream.copy().setMappings(mappingsOverrides).build();
     }