Add enrich source field mapper.

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/enrich/EnrichPolicy.java

@@ -29,7 +29,7 @@
  */
 public final class EnrichPolicy implements Writeable, ToXContentFragment {
 
-    private static final String ENRICH_INDEX_NAME_BASE = ".enrich-";
+    public static final String ENRICH_INDEX_NAME_BASE = ".enrich-";
 
     public static final String EXACT_MATCH_TYPE = "exact_match";
     public static final String[] SUPPORTED_POLICY_TYPES = new String[]{EXACT_MATCH_TYPE};

x-pack/plugin/enrich/qa/rest/src/test/java/org/elasticsearch/xpack/enrich/EnrichIT.java

@@ -5,19 +5,30 @@
  */
 package org.elasticsearch.xpack.enrich;
 
+import org.apache.http.entity.ByteArrayEntity;
+import org.apache.http.entity.ContentType;
 import org.apache.http.util.EntityUtils;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.test.rest.ESRestTestCase;
+import org.elasticsearch.test.rest.yaml.ObjectPath;
 import org.junit.After;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.util.Base64;
 import java.util.List;
 import java.util.Map;
 
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.notNullValue;
 
 public class EnrichIT extends ESRestTestCase {
 
@@ -76,6 +87,42 @@ public void testBasicFlow() throws Exception {
         assertThat(_source.get("tld_rank"), equalTo(7));
     }
 
+    public void testEnrichSourceMapping() throws Exception {
+        String indexName = ".enrich-1";
+        String mapping = "\"dynamic\": false,\"_source\": {\"enabled\": false},\"_enrich_source\": {\"enabled\": true}";
+        createIndex(indexName, Settings.EMPTY, mapping);
+
+        Request indexRequest = new Request("PUT", "/" + indexName + "/_doc/elastic.co");
+        XContentBuilder document = XContentBuilder.builder(XContentType.SMILE.xContent());
+        document.startObject();
+        document.field("globalRank", 25);
+        document.field("tldRank", 7);
+        document.field("tld", "co");
+        document.endObject();
+        document.close();
+        ByteArrayOutputStream out  = (ByteArrayOutputStream) document.getOutputStream();
+        indexRequest.setEntity(new ByteArrayEntity(out.toByteArray(), ContentType.create("application/smile")));
+        assertOK(client().performRequest(indexRequest));
+
+        Request refreshRequest = new Request("POST", "/" + indexName +"/_refresh");
+        assertOK(client().performRequest(refreshRequest));
+
+        Request searchRequest = new Request("GET", "/" + indexName +"/_search");
+        searchRequest.setJsonEntity("{\"docvalue_fields\": [{\"field\": \"_enrich_source\"}]}");
+        Map<String, ?> response = toMap(client().performRequest(searchRequest));
+        logger.info("RSP={}", response);
+        String enrichSource = ObjectPath.evaluate(response, "hits.hits.0.fields._enrich_source.0");
+        assertThat(enrichSource, notNullValue());
+
+        try (InputStream in = new ByteArrayInputStream(Base64.getDecoder().decode(enrichSource))) {
+            Map<String, Object> map = XContentHelper.convertToMap(XContentType.SMILE.xContent(), in, false);
+            assertThat(map.size(), equalTo(3));
+            assertThat(map.get("globalRank"), equalTo(25));
+            assertThat(map.get("tldRank"), equalTo(7));
+            assertThat(map.get("tld"), equalTo("co"));
+        }
+    }
+
     private static Map<String, Object> toMap(Response response) throws IOException {
         return toMap(EntityUtils.toString(response.getEntity()));
     }

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichPlugin.java

@@ -22,9 +22,11 @@
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.index.mapper.MetadataFieldMapper;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.IngestPlugin;
+import org.elasticsearch.plugins.MapperPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestHandler;
@@ -57,7 +59,7 @@
 import static java.util.Collections.emptyList;
 import static org.elasticsearch.xpack.core.XPackSettings.ENRICH_ENABLED_SETTING;
 
-public class EnrichPlugin extends Plugin implements ActionPlugin, IngestPlugin {
+public class EnrichPlugin extends Plugin implements ActionPlugin, IngestPlugin, MapperPlugin {
 
     static final Setting<Integer> ENRICH_FETCH_SIZE_SETTING =
         Setting.intSetting("index.xpack.enrich.fetch_size", 10000, 1, 1000000, Setting.Property.NodeScope);
@@ -70,6 +72,11 @@ public EnrichPlugin(final Settings settings) {
         this.enabled = ENRICH_ENABLED_SETTING.get(settings);
     }
 
+    @Override
+    public Map<String, MetadataFieldMapper.TypeParser> getMetadataMappers() {
+        return Map.of(EnrichSourceFieldMapper.NAME, new EnrichSourceFieldMapper.TypeParser());
+    }
+
     @Override
     public Map<String, Processor.Factory> getProcessors(Processor.Parameters parameters) {
         EnrichProcessorFactory factory = new EnrichProcessorFactory(parameters.localShardSearcher);

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichPolicyRunner.java

@@ -141,6 +141,9 @@ private XContentBuilder resolveEnrichMapping(final EnrichPolicy policy) {
                 .startObject(MapperService.SINGLE_MAPPING_NAME)
                     .field("dynamic", false)
                     .startObject("_source")
+                        .field("enabled", false)
+                    .endObject()
+                    .startObject("_enrich_source")
                         .field("enabled", true)
                     .endObject()
                     .startObject("properties")