Allow specify dynamic templates in bulk request

rest-api-spec/src/main/resources/rest-api-spec/test/bulk/11_type_hints.yml

@@ -0,0 +1,46 @@
+---
+"Match mapping type":
+  - skip:
+      version: " - 7.99.99"
+      reason: "introduced in 8.0"
+
+  - do:
+      indices.create:
+        index: test_index
+        body:
+          mappings:
+            dynamic_templates:
+              - locations:
+                  custom_mapping_type: location_type
+                  mapping:
+                    type: geo_point
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - index:
+              _index: test_index
+              _id: id_1
+              match_mapping_types:
+                location: location_type
+          - location: [ -71.34, 41.12 ]
+          - index:
+              _index: test_index
+              _id: id_2
+              match_mapping_types:
+                location: location_type
+          - location: "41.12,-71.34"
+  - do:
+      search:
+        index: test_index
+        body:
+          query:
+            geo_bounding_box:
+              location:
+                top_left:
+                  lat: 42
+                  lon: -72
+                bottom_right:
+                  lat: 40
+                  lon: -74
+  - match: { hits.total.value: 2 }

server/src/internalClusterTest/java/org/elasticsearch/index/mapper/DynamicMappingIT.java

@@ -9,28 +9,46 @@
 
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.metadata.MappingMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.Randomness;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.geo.GeoPoint;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.ObjectPath;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.index.query.GeoBoundingBoxQueryBuilder;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.InternalSettingsPlugin;
 import org.hamcrest.Matchers;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.elasticsearch.index.mapper.MapperService.INDEX_MAPPING_TOTAL_FIELDS_LIMIT_SETTING;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertSearchHits;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
 
 public class DynamicMappingIT extends ESIntegTestCase {
 
@@ -77,6 +95,14 @@ private static void assertMappingsHaveField(GetMappingsResponse mappings, String
         assertTrue("Could not find [" + field + "] in " + typeMappingsMap.toString(), properties.containsKey(field));
     }
 
+    private static void assertFieldMappingType(String index, String field, String expectedType) {
+        final GetMappingsResponse resp = client().admin().indices().prepareGetMappings(index).get();
+        final Map<String, Object> mappings = resp.mappings().get(index).sourceAsMap();
+        final String path = "properties." + String.join(".properties.", field.split("\\.")) + ".type";
+        Object actualType = ObjectPath.eval(path, mappings);
+        assertThat(Strings.toString(resp), actualType, equalTo(expectedType));
+    }
+
     public void testConcurrentDynamicUpdates() throws Throwable {
         createIndex("index");
         final Thread[] indexThreads = new Thread[32];
@@ -156,4 +182,76 @@ public void testMappingVersionAfterDynamicMappingUpdate() throws Exception {
         client().prepareIndex("test").setId("1").setSource("field", "text").get();
         assertBusy(() -> assertThat(clusterService.state().metadata().index("test").getMappingVersion(), equalTo(1 + previousVersion)));
     }
+
+    public void testSimpleTypeHint() throws Exception {
+        final XContentBuilder mappings = XContentFactory.jsonBuilder();
+        mappings.startObject();
+        {
+            mappings.startArray("dynamic_templates");
+            {
+                mappings.startObject();
+                mappings.startObject("locations");
+                {
+                    mappings.field("custom_mapping_type", "location");
+                    mappings.startObject("mapping");
+                    {
+                        mappings.field("type", "geo_point");
+                    }
+                    mappings.endObject();
+                }
+                mappings.endObject();
+                mappings.endObject();
+            }
+            mappings.endArray();
+        }
+        mappings.endObject();
+        assertAcked(client().admin().indices().prepareCreate("test").setMapping(mappings));
+        List<IndexRequest> requests = new ArrayList<>();
+        requests.add(new IndexRequest("test").id("1").source("location", "41.12,-71.34")
+            .setDynamicMappingTypeHints(Map.of("location", "location")));
+        requests.add(new IndexRequest("test").id("2").source(
+            XContentFactory.jsonBuilder()
+                .startObject()
+                .startObject("location").field("lat", 41.12).field("lon", -71.34).endObject()
+                .endObject())
+            .setDynamicMappingTypeHints(Map.of("location", "location")));
+        requests.add(new IndexRequest("test").id("3").source("address.location", "41.12,-71.34")
+            .setDynamicMappingTypeHints(Map.of("address.location", "location")));
+        requests.add(new IndexRequest("test").id("4").source("location", new double[]{-71.34, 41.12})
+            .setDynamicMappingTypeHints(Map.of("location", "location")));
+        requests.add(new IndexRequest("test").id("5").source("array_of_numbers", new double[]{-71.34, 41.12}));
+
+        Randomness.shuffle(requests);
+        BulkRequest bulkRequest = new BulkRequest().setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+        requests.forEach(bulkRequest::add);
+        client().bulk(bulkRequest).actionGet();
+
+        SearchResponse searchResponse = client().prepareSearch("test")
+            .setQuery(new GeoBoundingBoxQueryBuilder("location").setCorners(new GeoPoint(42, -72), new GeoPoint(40, -74)))
+            .get();
+        assertSearchHits(searchResponse, "1", "2", "4");
+        searchResponse = client().prepareSearch("test")
+            .setQuery(new GeoBoundingBoxQueryBuilder("address.location").setCorners(new GeoPoint(42, -72), new GeoPoint(40, -74)))
+            .get();
+        assertSearchHits(searchResponse, "3");
+    }
+
+    public void testTypeHintNotFound() throws Exception {
+        assertAcked(client().admin().indices().prepareCreate("test"));
+        BulkRequest bulkRequest = new BulkRequest().setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+        bulkRequest.add(
+            new IndexRequest("test").id("1").source(
+                XContentFactory.jsonBuilder()
+                    .startObject()
+                    .field("my_location",  "41.12,-71.34")
+                    .endObject())
+                .setDynamicMappingTypeHints(Map.of("my_location", "location"))
+        );
+        final BulkResponse bulkItemResponses = client().bulk(bulkRequest).actionGet();
+        assertTrue(bulkItemResponses.hasFailures());
+        for (BulkItemResponse resp : bulkItemResponses.getItems()) {
+            assertThat(resp.getFailure().getCause(), instanceOf(MapperParsingException.class));
+            assertThat(resp.getFailureMessage(), containsString("Can't find template for matching type [location] of field [my_location]"));
+        }
+    }
 }

server/src/main/java/org/elasticsearch/action/bulk/BulkRequestParser.java

@@ -53,6 +53,7 @@ public final class BulkRequestParser {
     private static final ParseField IF_SEQ_NO = new ParseField("if_seq_no");
     private static final ParseField IF_PRIMARY_TERM = new ParseField("if_primary_term");
     private static final ParseField REQUIRE_ALIAS = new ParseField(DocWriteRequest.REQUIRE_ALIAS);
+    private static final ParseField MATCH_MAPPING_TYPES = new ParseField("match_mapping_types");
 
     // TODO: Remove this parameter once the BulkMonitoring endpoint has been removed
     private final boolean errorOnType;
@@ -156,6 +157,7 @@ public void parse(
                 int retryOnConflict = 0;
                 String pipeline = defaultPipeline;
                 boolean requireAlias = defaultRequireAlias != null && defaultRequireAlias;
+                Map<String, String> matchMappingTypes = Map.of();
 
                 // at this stage, next token can either be END_OBJECT (and use default index and type, with auto generated id)
                 // or START_OBJECT which will have another set of parameters
@@ -206,7 +208,10 @@ public void parse(
                             }
                         } else if (token == XContentParser.Token.START_ARRAY) {
                             throw new IllegalArgumentException("Malformed action/metadata line [" + line +
-                                    "], expected a simple value for field [" + currentFieldName + "] but found [" + token + "]");
+                                "], expected a simple value for field [" + currentFieldName + "] but found [" + token + "]");
+                        } else if (token == XContentParser.Token.START_OBJECT &&
+                            MATCH_MAPPING_TYPES.match(currentFieldName, parser.getDeprecationHandler())) {
+                            matchMappingTypes = parser.mapStrings();
                         } else if (token == XContentParser.Token.START_OBJECT && SOURCE.match(currentFieldName,
                                 parser.getDeprecationHandler())) {
                             fetchSourceContext = FetchSourceContext.fromXContent(parser);
@@ -238,6 +243,7 @@ public void parse(
                                     .version(version).versionType(versionType)
                                     .setPipeline(pipeline).setIfSeqNo(ifSeqNo).setIfPrimaryTerm(ifPrimaryTerm)
                                     .source(sliceTrimmingCarriageReturn(data, from, nextMarker, xContentType), xContentType)
+                                    .setDynamicMappingTypeHints(matchMappingTypes)
                                     .setRequireAlias(requireAlias), type);
                         } else {
                             indexRequestConsumer.accept(new IndexRequest(index).id(id).routing(routing)
@@ -252,6 +258,7 @@ public void parse(
                                 .version(version).versionType(versionType)
                                 .create(true).setPipeline(pipeline).setIfSeqNo(ifSeqNo).setIfPrimaryTerm(ifPrimaryTerm)
                                 .source(sliceTrimmingCarriageReturn(data, from, nextMarker, xContentType), xContentType)
+                                .setDynamicMappingTypeHints(matchMappingTypes)
                                 .setRequireAlias(requireAlias), type);
                     } else if ("update".equals(action)) {
                         if (version != Versions.MATCH_ANY || versionType != VersionType.INTERNAL) {

server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -281,8 +281,9 @@ static boolean executeBulkItemRequest(BulkPrimaryExecutionContext context, Updat
                 request.ifSeqNo(), request.ifPrimaryTerm());
         } else {
             final IndexRequest request = context.getRequestToExecute();
-            result = primary.applyIndexOperationOnPrimary(version, request.versionType(), new SourceToParse(
-                    request.index(), request.id(), request.source(), request.getContentType(), request.routing()),
+            final SourceToParse sourceToParse = new SourceToParse(request.index(), request.id(), request.source(),
+                request.getContentType(), request.routing(), request.getDynamicMappingTypeHints());
+            result = primary.applyIndexOperationOnPrimary(version, request.versionType(), sourceToParse,
                     request.ifSeqNo(), request.ifPrimaryTerm(), request.getAutoGeneratedTimestamp(), request.isRetry());
         }
         if (result.getResultType() == Engine.Result.Type.MAPPING_UPDATE_REQUIRED) {
@@ -479,8 +480,8 @@ private static Engine.Result performOpOnReplica(DocWriteResponse primaryResponse
             case INDEX:
                 final IndexRequest indexRequest = (IndexRequest) docWriteRequest;
                 final ShardId shardId = replica.shardId();
-                final SourceToParse sourceToParse = new SourceToParse(shardId.getIndexName(), indexRequest.id(),
-                    indexRequest.source(), indexRequest.getContentType(), indexRequest.routing());
+                final SourceToParse sourceToParse = new SourceToParse(shardId.getIndexName(), indexRequest.id(), indexRequest.source(),
+                    indexRequest.getContentType(), indexRequest.routing(), Map.of());
                 result = replica.applyIndexOperationOnReplica(primaryResponse.getSeqNo(), primaryResponse.getPrimaryTerm(),
                     primaryResponse.getVersion(), indexRequest.getAutoGeneratedTimestamp(), indexRequest.isRetry(), sourceToParse);
                 break;

server/src/main/java/org/elasticsearch/action/index/IndexRequest.java

@@ -109,6 +109,8 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
     private long ifSeqNo = UNASSIGNED_SEQ_NO;
     private long ifPrimaryTerm = UNASSIGNED_PRIMARY_TERM;
 
+    private Map<String, String> dynamicMappingTypeHints = Map.of();
+
     public IndexRequest(StreamInput in) throws IOException {
         this(null, in);
     }
@@ -146,6 +148,11 @@ public IndexRequest(@Nullable ShardId shardId, StreamInput in) throws IOExceptio
         } else {
             requireAlias = false;
         }
+        if (in.getVersion().onOrAfter(Version.V_8_0_0)) {
+            dynamicMappingTypeHints = in.readMap(StreamInput::readString, StreamInput::readString);
+        } else {
+            dynamicMappingTypeHints = Map.of();
+        }
     }
 
     public IndexRequest() {
@@ -655,6 +662,13 @@ private void writeBody(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_7_10_0)) {
             out.writeBoolean(requireAlias);
         }
+        if (out.getVersion().onOrAfter(Version.V_8_0_0)) {
+            out.writeMap(dynamicMappingTypeHints, StreamOutput::writeString, StreamOutput::writeString);
+        } else {
+            if (dynamicMappingTypeHints.isEmpty() == false) {
+                throw new IllegalStateException("Dynamic mapping type hints requires all nodes in the cluster on 8.0 or later");
+            }
+        }
     }
 
     @Override
@@ -712,4 +726,13 @@ public IndexRequest setRequireAlias(boolean requireAlias) {
         this.requireAlias = requireAlias;
         return this;
     }
+
+    public IndexRequest setDynamicMappingTypeHints(Map<String, String> dynamicMappingTypeHints) {
+        this.dynamicMappingTypeHints = Objects.requireNonNull(dynamicMappingTypeHints);
+        return this;
+    }
+
+    public Map<String, String> getDynamicMappingTypeHints() {
+        return dynamicMappingTypeHints;
+    }
 }

server/src/main/java/org/elasticsearch/index/engine/SingleDocDirectoryReader.java

@@ -37,6 +37,7 @@
 import org.elasticsearch.index.translog.Translog;
 
 import java.io.IOException;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
@@ -140,7 +141,8 @@ private LeafReader getDelegate() {
         private LeafReader createInMemoryLeafReader() {
             assert Thread.holdsLock(this);
             final ParsedDocument parsedDocs = mapper.parse(new SourceToParse(shardId.getIndexName(), operation.id(),
-                operation.source(), XContentHelper.xContentType(operation.source()), operation.routing()));
+                operation.source(), XContentHelper.xContentType(operation.source()), operation.routing(), Map.of()));
+
             parsedDocs.updateSeqID(operation.seqNo(), operation.primaryTerm());
             parsedDocs.version().setLongValue(operation.version());
             final IndexWriterConfig writeConfig = new IndexWriterConfig(analyzer).setOpenMode(IndexWriterConfig.OpenMode.CREATE);

server/src/main/java/org/elasticsearch/index/get/ShardGetService.java

@@ -224,7 +224,7 @@ private GetResult innerGetLoadFromStoredFields(String id, String[] storedFields,
                     // Slow path: recreate stored fields from original source
                     assert source != null : "original source in translog must exist";
                     SourceToParse sourceToParse = new SourceToParse(shardId.getIndexName(), id, source, XContentHelper.xContentType(source),
-                        fieldVisitor.routing());
+                        fieldVisitor.routing(), Map.of());
                     ParsedDocument doc = indexShard.mapperService().documentMapper().parse(sourceToParse);
                     assert doc.dynamicMappingsUpdate() == null : "mapping updates should not be required on already-indexed doc";
                     // update special fields

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

@@ -543,7 +543,8 @@ private static void parseArray(ParseContext context, ObjectMapper parentMapper,
                 // TODO: shouldn't this skip, not parse?
                 parseNonDynamicArray(context, parentMapper, lastFieldName, arrayFieldName);
             } else {
-                Mapper objectMapperFromTemplate = dynamic.getDynamicFieldsBuilder().createObjectMapperFromTemplate(context, arrayFieldName);
+                Mapper objectMapperFromTemplate =
+                    dynamic.getDynamicFieldsBuilder().createFieldOrObjectMapperFromTemplate(context, arrayFieldName);
                 if (objectMapperFromTemplate == null) {
                     parseNonDynamicArray(context, parentMapper, lastFieldName, arrayFieldName);
                 } else {
@@ -711,7 +712,12 @@ private static Tuple<Integer, ObjectMapper> getDynamicParentMapper(ParseContext
                     return new Tuple<>(pathsAdded, parent);
                 } else {
                     //objects are created under properties even with dynamic: runtime, as the runtime section only holds leaf fields
-                    mapper = (ObjectMapper) dynamic.getDynamicFieldsBuilder().createDynamicObjectMapper(context, paths[i]);
+                    final Mapper fieldMapper = dynamic.getDynamicFieldsBuilder().createDynamicObjectMapper(context, paths[i]);
+                    if (fieldMapper instanceof ObjectMapper == false) {
+                        throw new MapperParsingException("Field [" + context.path().pathAsText(paths[i]) + "] must be an object; " +
+                            "but it's configured as [" + fieldMapper.typeName() + "] in dynamic templates");
+                    }
+                    mapper = (ObjectMapper) fieldMapper;
                     if (mapper.nested() != ObjectMapper.Nested.NO) {
                         throw new MapperParsingException("It is forbidden to create dynamic nested objects (["
                             + context.path().pathAsText(paths[i]) + "]) through `copy_to` or dots in field names");