Skip to content

Commit 1868594

Browse files
masseykemasseyke
authored
Computing full effective mappings for data stream mapping APIs (elastic#130498)
1 parent 818d31a commit 1868594

File tree

10 files changed

+304
-34
lines changed

10 files changed

+304
-34
lines changed

‎modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamIT.java‎

Lines changed: 164 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import org.elasticsearch.ElasticsearchStatusException;
1313
import org.elasticsearch.ExceptionsHelper;
1414
import org.elasticsearch.action.ActionFuture;
15+
import org.elasticsearch.action.ActionType;
1516
import org.elasticsearch.action.DocWriteRequest;
1617
import org.elasticsearch.action.DocWriteResponse;
1718
import org.elasticsearch.action.RequestBuilder;
@@ -41,6 +42,7 @@
4142
import org.elasticsearch.action.admin.indices.stats.ShardStats;
4243
import org.elasticsearch.action.admin.indices.template.delete.TransportDeleteComposableIndexTemplateAction;
4344
import org.elasticsearch.action.admin.indices.template.get.GetComposableIndexTemplateAction;
45+
import org.elasticsearch.action.admin.indices.template.put.PutComponentTemplateAction;
4446
import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateRequest;
4547
import org.elasticsearch.action.admin.indices.template.put.TransportPutComposableIndexTemplateAction;
4648
import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryRequestBuilder;
@@ -52,6 +54,7 @@
5254
import org.elasticsearch.action.datastreams.GetDataStreamAction;
5355
import org.elasticsearch.action.datastreams.GetDataStreamAction.Response.DataStreamInfo;
5456
import org.elasticsearch.action.datastreams.ModifyDataStreamsAction;
57+
import org.elasticsearch.action.datastreams.UpdateDataStreamMappingsAction;
5558
import org.elasticsearch.action.delete.DeleteRequest;
5659
import org.elasticsearch.action.index.IndexRequest;
5760
import org.elasticsearch.action.search.MultiSearchRequestBuilder;
@@ -62,6 +65,7 @@
6265
import org.elasticsearch.cluster.ClusterStateUpdateTask;
6366
import org.elasticsearch.cluster.health.ClusterHealthStatus;
6467
import org.elasticsearch.cluster.metadata.AliasMetadata;
68+
import org.elasticsearch.cluster.metadata.ComponentTemplate;
6569
import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
6670
import org.elasticsearch.cluster.metadata.DataStream;
6771
import org.elasticsearch.cluster.metadata.DataStreamAction;
@@ -72,20 +76,24 @@
7276
import org.elasticsearch.cluster.metadata.IndexMetadataStats;
7377
import org.elasticsearch.cluster.metadata.IndexWriteLoad;
7478
import org.elasticsearch.cluster.metadata.Metadata;
79+
import org.elasticsearch.cluster.metadata.ProjectId;
80+
import org.elasticsearch.cluster.metadata.ProjectMetadata;
7581
import org.elasticsearch.cluster.metadata.Template;
7682
import org.elasticsearch.cluster.routing.IndexRoutingTable;
7783
import org.elasticsearch.cluster.routing.ShardRouting;
7884
import org.elasticsearch.cluster.service.ClusterService;
7985
import org.elasticsearch.common.Strings;
8086
import org.elasticsearch.common.compress.CompressedXContent;
8187
import org.elasticsearch.common.settings.Settings;
88+
import org.elasticsearch.common.xcontent.XContentHelper;
8289
import org.elasticsearch.core.Nullable;
8390
import org.elasticsearch.index.Index;
8491
import org.elasticsearch.index.IndexNotFoundException;
8592
import org.elasticsearch.index.mapper.DataStreamTimestampFieldMapper;
8693
import org.elasticsearch.index.mapper.DateFieldMapper;
8794
import org.elasticsearch.index.query.TermQueryBuilder;
8895
import org.elasticsearch.index.shard.IndexingStats;
96+
import org.elasticsearch.indices.IndicesService;
8997
import org.elasticsearch.indices.InvalidAliasNameException;
9098
import org.elasticsearch.indices.InvalidIndexNameException;
9199
import org.elasticsearch.plugins.Plugin;
@@ -2411,6 +2419,162 @@ public void testShardSizeIsForecastedDuringRollover() throws Exception {
24112419
assertThat(forecastedShardSizeInBytes.getAsLong(), is(equalTo(expectedTotalSizeInBytes / shardCount)));
24122420
}
24132421

2422+
public void testGetEffectiveMappings() throws Exception {
2423+
/*
2424+
* This test creates a composable template with a mapping and with two component templates with mappings. It then makes sure that
2425+
* DataStream.getEffectiveMappings returns a mapping that merges the template's mapping, the component templates' mappings, and the
2426+
* mapping override given. It then makes sure we get the same result calling the non-static version of getEffectiveMappings.
2427+
*/
2428+
ComposableIndexTemplate composableIndexTemplate;
2429+
{
2430+
ComponentTemplate ct1 = new ComponentTemplate(new Template(null, new CompressedXContent("""
2431+
{
2432+
"_doc":{
2433+
"dynamic":"strict",
2434+
"properties":{
2435+
"field1":{
2436+
"type":"text"
2437+
}
2438+
}
2439+
}
2440+
}
2441+
"""), null), 3L, null);
2442+
ComponentTemplate ct2 = new ComponentTemplate(new Template(null, new CompressedXContent("""
2443+
{
2444+
"_doc":{
2445+
"dynamic":"strict",
2446+
"properties":{
2447+
"field2":{
2448+
"type":"text"
2449+
}
2450+
}
2451+
}
2452+
}
2453+
"""), null), 3L, null);
2454+
client().execute(PutComponentTemplateAction.INSTANCE, new PutComponentTemplateAction.Request("ct1").componentTemplate(ct1))
2455+
.get();
2456+
client().execute(PutComponentTemplateAction.INSTANCE, new PutComponentTemplateAction.Request("ct2").componentTemplate(ct2))
2457+
.get();
2458+
2459+
List<String> componentTemplates = List.of("ct1", "ct2");
2460+
String templateName = "effective-mapping-template";
2461+
TransportPutComposableIndexTemplateAction.Request request = new TransportPutComposableIndexTemplateAction.Request(templateName);
2462+
request.indexTemplate(
2463+
ComposableIndexTemplate.builder()
2464+
.indexPatterns(List.of("effective-*"))
2465+
.template(Template.builder().mappings(CompressedXContent.fromJSON("""
2466+
{
2467+
"_doc":{
2468+
"dynamic":"strict",
2469+
"properties":{
2470+
"field3":{
2471+
"type":"text"
2472+
}
2473+
}
2474+
}
2475+
}
2476+
""")))
2477+
.componentTemplates(componentTemplates)
2478+
.dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
2479+
.build()
2480+
);
2481+
client().execute(TransportPutComposableIndexTemplateAction.TYPE, request).actionGet();
2482+
GetComposableIndexTemplateAction.Response getTemplateResponse = client().execute(
2483+
GetComposableIndexTemplateAction.INSTANCE,
2484+
new GetComposableIndexTemplateAction.Request(TEST_REQUEST_TIMEOUT, templateName)
2485+
).actionGet();
2486+
composableIndexTemplate = getTemplateResponse.indexTemplates().values().iterator().next();
2487+
}
2488+
// The mappingOverrides changes the type of one field, and adds another field:
2489+
CompressedXContent mappingOverrides = new CompressedXContent("""
2490+
{
2491+
"properties":{
2492+
"field1":{
2493+
"type":"keyword"
2494+
},
2495+
"field4":{
2496+
"type":"keyword"
2497+
}
2498+
}
2499+
}
2500+
""");
2501+
2502+
String dataStreamName = "effective-mappings-test";
2503+
Index writeIndex;
2504+
{
2505+
CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(
2506+
TEST_REQUEST_TIMEOUT,
2507+
TEST_REQUEST_TIMEOUT,
2508+
dataStreamName
2509+
);
2510+
client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).get();
2511+
writeIndex = getDataStream(dataStreamName).getWriteIndex();
2512+
}
2513+
2514+
ProjectMetadata projectMetadata = client().admin()
2515+
.cluster()
2516+
.state(new ClusterStateRequest(TEST_REQUEST_TIMEOUT))
2517+
.get()
2518+
.getState()
2519+
.metadata()
2520+
.getProject(ProjectId.DEFAULT);
2521+
IndicesService indicesService = internalCluster().getInstance(IndicesService.class);
2522+
CompressedXContent effectiveMappings = DataStream.getEffectiveMappings(
2523+
projectMetadata,
2524+
composableIndexTemplate,
2525+
mappingOverrides,
2526+
writeIndex,
2527+
indicesService
2528+
);
2529+
assertNotNull(effectiveMappings);
2530+
Map<String, Object> effectiveMappingMap = XContentHelper.convertToMap(effectiveMappings.uncompressed(), true, XContentType.JSON)
2531+
.v2();
2532+
Map<String, Object> expectedEffectiveMappingMap = Map.of(
2533+
"_doc",
2534+
Map.of(
2535+
"dynamic",
2536+
"strict",
2537+
"_data_stream_timestamp",
2538+
Map.of("enabled", true),
2539+
"properties",
2540+
Map.of(
2541+
"@timestamp",
2542+
Map.of("type", "date"),
2543+
"field1",
2544+
Map.of("type", "keyword"),
2545+
"field2",
2546+
Map.of("type", "text"),
2547+
"field3",
2548+
Map.of("type", "text"),
2549+
"field4",
2550+
Map.of("type", "keyword")
2551+
)
2552+
)
2553+
);
2554+
assertThat(effectiveMappingMap, equalTo(expectedEffectiveMappingMap));
2555+
2556+
// Add the same mappingOverrides to the data stream:
2557+
client().execute(
2558+
new ActionType<UpdateDataStreamMappingsAction.Response>(UpdateDataStreamMappingsAction.NAME),
2559+
new UpdateDataStreamMappingsAction.Request(mappingOverrides, false, TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT).indices(
2560+
dataStreamName
2561+
)
2562+
).actionGet();
2563+
assertThat(getDataStream(dataStreamName).getEffectiveMappings(projectMetadata, indicesService), equalTo(effectiveMappings));
2564+
}
2565+
2566+
private DataStream getDataStream(String dataStreamName) throws ExecutionException, InterruptedException {
2567+
return client().admin()
2568+
.cluster()
2569+
.state(new ClusterStateRequest(TEST_REQUEST_TIMEOUT))
2570+
.get()
2571+
.getState()
2572+
.getMetadata()
2573+
.getProject(ProjectId.DEFAULT)
2574+
.dataStreams()
2575+
.get(dataStreamName);
2576+
}
2577+
24142578
private void indexDocsAndEnsureThereIsCapturedWriteLoad(String dataStreamName) throws Exception {
24152579
assertBusy(() -> {
24162580
for (int i = 0; i < 10; i++) {

‎modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/TransportUpdateDataStreamMappingsActionIT.java‎

Lines changed: 27 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -55,20 +55,39 @@ public void testGetAndUpdateMappings() throws IOException {
5555
createDataStream(dataStreamName);
5656

5757
Map<String, Object> originalMappings = Map.of(
58-
"dynamic",
59-
"strict",
60-
"properties",
61-
Map.of("foo1", Map.of("type", "text"), "foo2", Map.of("type", "text"))
58+
"_doc",
59+
Map.of(
60+
"dynamic",
61+
"strict",
62+
"_data_stream_timestamp",
63+
Map.of("enabled", true),
64+
"properties",
65+
Map.of("@timestamp", Map.of("type", "date"), "foo1", Map.of("type", "text"), "foo2", Map.of("type", "text"))
66+
)
6267
);
6368
Map<String, Object> mappingOverrides = Map.of(
6469
"properties",
6570
Map.of("foo2", Map.of("type", "keyword"), "foo3", Map.of("type", "text"))
6671
);
6772
Map<String, Object> expectedEffectiveMappings = Map.of(
68-
"dynamic",
69-
"strict",
70-
"properties",
71-
Map.of("foo1", Map.of("type", "text"), "foo2", Map.of("type", "keyword"), "foo3", Map.of("type", "text"))
73+
"_doc",
74+
Map.of(
75+
"dynamic",
76+
"strict",
77+
"_data_stream_timestamp",
78+
Map.of("enabled", true),
79+
"properties",
80+
Map.of(
81+
"@timestamp",
82+
Map.of("type", "date"),
83+
"foo1",
84+
Map.of("type", "text"),
85+
"foo2",
86+
Map.of("type", "keyword"),
87+
"foo3",
88+
Map.of("type", "text")
89+
)
90+
)
7291
);
7392
assertExpectedMappings(dataStreamName, Map.of(), originalMappings);
7493
updateMappings(dataStreamName, mappingOverrides, expectedEffectiveMappings, true);

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

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
2323
import org.elasticsearch.cluster.project.ProjectResolver;
2424
import org.elasticsearch.cluster.service.ClusterService;
25+
import org.elasticsearch.indices.IndicesService;
2526
import org.elasticsearch.injection.guice.Inject;
2627
import org.elasticsearch.tasks.Task;
2728
import org.elasticsearch.threadpool.ThreadPool;
@@ -35,6 +36,7 @@ public class TransportGetDataStreamMappingsAction extends TransportLocalProjectM
3536
GetDataStreamMappingsAction.Request,
3637
GetDataStreamMappingsAction.Response> {
3738
private final IndexNameExpressionResolver indexNameExpressionResolver;
39+
private final IndicesService indicesService;
3840

3941
@Inject
4042
public TransportGetDataStreamMappingsAction(
@@ -43,7 +45,8 @@ public TransportGetDataStreamMappingsAction(
4345
ThreadPool threadPool,
4446
ActionFilters actionFilters,
4547
ProjectResolver projectResolver,
46-
IndexNameExpressionResolver indexNameExpressionResolver
48+
IndexNameExpressionResolver indexNameExpressionResolver,
49+
IndicesService indicesService
4750
) {
4851
super(
4952
GetSettingsAction.NAME,
@@ -54,6 +57,7 @@ public TransportGetDataStreamMappingsAction(
5457
projectResolver
5558
);
5659
this.indexNameExpressionResolver = indexNameExpressionResolver;
60+
this.indicesService = indicesService;
5761
}
5862

5963
@Override
@@ -81,7 +85,7 @@ protected void localClusterStateOperation(
8185
new GetDataStreamMappingsAction.DataStreamMappingsResponse(
8286
dataStreamName,
8387
dataStream.getMappings(),
84-
dataStream.getEffectiveMappings(project.metadata())
88+
dataStream.getEffectiveMappings(project.metadata(), indicesService)
8589
)
8690
);
8791
}

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

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import org.elasticsearch.common.util.concurrent.EsExecutors;
3030
import org.elasticsearch.core.TimeValue;
3131
import org.elasticsearch.index.mapper.Mapping;
32+
import org.elasticsearch.indices.IndicesService;
3233
import org.elasticsearch.indices.SystemIndices;
3334
import org.elasticsearch.injection.guice.Inject;
3435
import org.elasticsearch.tasks.Task;
@@ -47,6 +48,7 @@ public class TransportUpdateDataStreamMappingsAction extends TransportMasterNode
4748
private final IndexNameExpressionResolver indexNameExpressionResolver;
4849
private final SystemIndices systemIndices;
4950
private final ProjectResolver projectResolver;
51+
private final IndicesService indicesService;
5052

5153
@Inject
5254
public TransportUpdateDataStreamMappingsAction(
@@ -57,7 +59,8 @@ public TransportUpdateDataStreamMappingsAction(
5759
ProjectResolver projectResolver,
5860
MetadataDataStreamsService metadataDataStreamsService,
5961
IndexNameExpressionResolver indexNameExpressionResolver,
60-
SystemIndices systemIndices
62+
SystemIndices systemIndices,
63+
IndicesService indicesService
6164
) {
6265
super(
6366
UpdateDataStreamMappingsAction.NAME,
@@ -73,6 +76,7 @@ public TransportUpdateDataStreamMappingsAction(
7376
this.metadataDataStreamsService = metadataDataStreamsService;
7477
this.indexNameExpressionResolver = indexNameExpressionResolver;
7578
this.systemIndices = systemIndices;
79+
this.indicesService = indicesService;
7680
}
7781

7882
@Override
@@ -163,7 +167,7 @@ private void updateSingleDataStream(
163167
true,
164168
null,
165169
mappingsOverrides,
166-
dataStream.getEffectiveMappings(clusterService.state().metadata().getProject(projectId))
170+
dataStream.getEffectiveMappings(clusterService.state().metadata().getProject(projectId), indicesService)
167171
)
168172
);
169173
} catch (IOException e) {

0 commit comments

Comments
 (0)