Geo-Match Enrich Processor

this commit introduces a geo-match enrich processor that looks up a specific
`geo_point` field in the enrich-index for all entries that have a geo_shape match field
that meets some specific relation criteria with the input field.

For example, the enrich index may contain documents with zipcodes and their respective
geo_shape. Ingesting documents with a geo_point field can be enriched with which zipcode
they associate according to which shape they are contained within.

this commit also refactors some of the MatchProcessor by moving a lot of the shared code to
AbstractEnrichProcessor.

Closes elastic#42639.

Author: talevy

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

@@ -5,18 +5,151 @@
  */
 package org.elasticsearch.xpack.enrich;
 
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.routing.Preference;
 import org.elasticsearch.ingest.AbstractProcessor;
+import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.xpack.core.enrich.EnrichPolicy;
+import org.elasticsearch.xpack.enrich.action.EnrichCoordinatorProxyAction;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiConsumer;
 
 public abstract class AbstractEnrichProcessor extends AbstractProcessor {
 
     private final String policyName;
+    private final BiConsumer<SearchRequest, BiConsumer<SearchResponse, Exception>> searchRunner;
+    private final String field;
+    private final String targetField;
+    private final boolean ignoreMissing;
+    private final boolean overrideEnabled;
+    protected final String matchField;
+    protected final int maxMatches;
+
+
+    protected AbstractEnrichProcessor(String tag, Client client, String policyName, String field, String targetField,
+                                      boolean ignoreMissing, boolean overrideEnabled, String matchField, int maxMatches) {
+        this(tag, createSearchRunner(client), policyName, field, targetField, ignoreMissing, overrideEnabled, matchField, maxMatches);
+    }
 
-    protected AbstractEnrichProcessor(String tag, String policyName) {
+    protected AbstractEnrichProcessor(String tag,
+                                      BiConsumer<SearchRequest, BiConsumer<SearchResponse, Exception>> searchRunner,
+                                      String policyName, String field, String targetField, boolean ignoreMissing, boolean overrideEnabled,
+                                      String matchField, int maxMatches) {
         super(tag);
         this.policyName = policyName;
+        this.searchRunner = searchRunner;
+        this.field = field;
+        this.targetField = targetField;
+        this.ignoreMissing = ignoreMissing;
+        this.overrideEnabled = overrideEnabled;
+        this.matchField = matchField;
+        this.maxMatches = maxMatches;
+    }
+
+    public abstract SearchSourceBuilder getSearchSourceBuilder(Object fieldValue);
+
+    @Override
+    public void execute(IngestDocument ingestDocument, BiConsumer<IngestDocument, Exception> handler) {
+        try {
+            // If a document does not have the enrich key, return the unchanged document
+            final Object value = ingestDocument.getFieldValue(field, Object.class, ignoreMissing);
+            if (value == null) {
+                handler.accept(ingestDocument, null);
+                return;
+            }
+
+            SearchSourceBuilder searchSourceBuilder = getSearchSourceBuilder(value);
+
+            SearchRequest req = new SearchRequest();
+            req.indices(EnrichPolicy.getBaseName(getPolicyName()));
+            req.preference(Preference.LOCAL.type());
+            req.source(searchSourceBuilder);
+
+            searchRunner.accept(req, (searchResponse, e) -> {
+                if (e != null) {
+                    handler.accept(null, e);
+                    return;
+                }
+
+                // If the index is empty, return the unchanged document
+                // If the enrich key does not exist in the index, throw an error
+                // If no documents match the key, return the unchanged document
+                SearchHit[] searchHits = searchResponse.getHits().getHits();
+                if (searchHits.length < 1) {
+                    handler.accept(ingestDocument, null);
+                    return;
+                }
+
+                if (overrideEnabled || ingestDocument.hasField(targetField) == false) {
+                    List<Map<String, Object>> enrichDocuments = new ArrayList<>(searchHits.length);
+                    for (SearchHit searchHit : searchHits) {
+                        Map<String, Object> enrichDocument = searchHit.getSourceAsMap();
+                        enrichDocuments.add(enrichDocument);
+                    }
+                    ingestDocument.setFieldValue(targetField, enrichDocuments);
+                }
+                handler.accept(ingestDocument, null);
+            });
+        } catch (Exception e) {
+            handler.accept(null, e);
+        }
+    }
+
+    @Override
+    public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
+        throw new UnsupportedOperationException("this method should not get executed");
     }
 
     public String getPolicyName() {
         return policyName;
     }
+
+    @Override
+    public String getType() {
+        return EnrichProcessorFactory.TYPE;
+    }
+
+    String getField() {
+        return field;
+    }
+
+    public String getTargetField() {
+        return targetField;
+    }
+
+    boolean isIgnoreMissing() {
+        return ignoreMissing;
+    }
+
+    boolean isOverrideEnabled() {
+        return overrideEnabled;
+    }
+
+    public String getMatchField() {
+        return matchField;
+    }
+
+    int getMaxMatches() {
+        return maxMatches;
+    }
+
+    private static BiConsumer<SearchRequest, BiConsumer<SearchResponse, Exception>> createSearchRunner(Client client) {
+        return (req, handler) -> {
+            client.execute(EnrichCoordinatorProxyAction.INSTANCE, req, ActionListener.wrap(
+                resp -> {
+                    handler.accept(resp, null);
+                },
+                e -> {
+                    handler.accept(null, e);
+                }));
+        };
+    }
 }

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

@@ -196,31 +196,29 @@ private void validateField(Map<?, ?> properties, String fieldName, boolean field
 
     private XContentBuilder resolveEnrichMapping(final EnrichPolicy policy) {
         // Currently the only supported policy type is EnrichPolicy.MATCH_TYPE, which is a keyword type
-        String keyType;
+        final String keyType;
+        final MatchFieldMapping matchFieldMapping;
         if (EnrichPolicy.MATCH_TYPE.equals(policy.getType())) {
-            keyType = "keyword";
+            matchFieldMapping = (builder) -> builder.field("type", "keyword").field("doc_values", false);
             // No need to also configure index_options, because keyword type defaults to 'docs'.
         } else if (EnrichPolicy.GEO_MATCH_TYPE.equals(policy.getType())) {
-            keyType = "geo_shape";
+            matchFieldMapping = (builder) -> builder.field("type", "geo_shape");
         } else {
             throw new ElasticsearchException("Unrecognized enrich policy type [{}]", policy.getType());
         }
 
         // Enable _source on enrich index. Explicitly mark key mapping type.
         try {
             XContentBuilder builder = JsonXContent.contentBuilder();
-            builder.startObject()
+            builder = builder.startObject()
                 .startObject(MapperService.SINGLE_MAPPING_NAME)
                     .field("dynamic", false)
                     .startObject("_source")
                         .field("enabled", true)
                     .endObject()
                     .startObject("properties")
-                        .startObject(policy.getMatchField())
-                            .field("type", keyType)
-                            .field("doc_values", false)
-                        .endObject()
-                    .endObject()
+                        .startObject(policy.getMatchField());
+            builder = matchFieldMapping.build(builder).endObject().endObject()
                     .startObject("_meta")
                         .field(ENRICH_README_FIELD_NAME, ENRICH_INDEX_README_TEXT)
                         .field(ENRICH_POLICY_NAME_FIELD_NAME, policyName)
@@ -413,4 +411,9 @@ public void onFailure(Exception e) {
             }
         });
     }
+
+    @FunctionalInterface
+    private interface MatchFieldMapping {
+        XContentBuilder build(XContentBuilder builder) throws IOException;
+    }
 }

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

@@ -10,6 +10,7 @@
 import org.elasticsearch.cluster.metadata.AliasOrIndex;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.common.geo.ShapeRelation;
 import org.elasticsearch.common.xcontent.support.XContentMapValues;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.Processor;
@@ -57,11 +58,11 @@ public Processor create(Map<String, Processor.Factory> processorFactories, Strin
 
         switch (policyType) {
             case EnrichPolicy.MATCH_TYPE:
-                return new MatchProcessor(tag, client, policyName, field, targetField, matchField,
-                    ignoreMissing, overrideEnabled, maxMatches);
+                return new MatchProcessor(tag, client, policyName, field, targetField, overrideEnabled, ignoreMissing, matchField,
+                    maxMatches);
             case EnrichPolicy.GEO_MATCH_TYPE:
-                return new GeoMatchProcessor(tag, client, policyName, field, targetField, matchField,
-                    ignoreMissing, overrideEnabled, maxMatches);
+                return new GeoMatchProcessor(tag, client, policyName, field, targetField, overrideEnabled, ignoreMissing, matchField,
+                    maxMatches, ShapeRelation.INTERSECTS);
             default:
                 throw new IllegalArgumentException("unsupported policy type [" + policyType + "]");
         }