match processor should handler values other than string properly (#47419)

Currently if the document being ingested contains another field value
than a string then the processor fails with an error.

This commit changes the match processor to handle number values
and array values correctly.

If a json array is detected then the `terms` query is used instead
of the `term` query.

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

@@ -10,7 +10,9 @@ import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.TermQueryBuilder;
+import org.elasticsearch.index.query.TermsQueryBuilder;
 
+import java.util.List;
 import java.util.function.BiConsumer;
 
 public class MatchProcessor extends AbstractEnrichProcessor {
@@ -42,6 +44,10 @@ public class MatchProcessor extends AbstractEnrichProcessor {
 
     @Override
     public QueryBuilder getQueryBuilder(Object fieldValue) {
-        return new TermQueryBuilder(matchField, fieldValue);
+        if (fieldValue instanceof List) {
+            return new TermsQueryBuilder(matchField, (List) fieldValue);
+        } else {
+            return new TermQueryBuilder(matchField, fieldValue);
+        }
     }
 }

x-pack/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/MatchProcessorTests.java

@@ -19,6 +19,7 @@ import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.query.ConstantScoreQueryBuilder;
 import org.elasticsearch.index.query.TermQueryBuilder;
+import org.elasticsearch.index.query.TermsQueryBuilder;
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.SearchHits;
@@ -204,6 +205,69 @@ public class MatchProcessorTests extends ESTestCase {
         assertThat(resultHolder[0].getFieldValue("tld", Object.class), equalTo(null));
     }
 
+    public void testNumericValue() {
+        int maxMatches = randomIntBetween(1, 8);
+        MockSearchFunction mockSearch = mockedSearchFunction(Map.of(2, Map.of("globalRank", 451, "tldRank", 23, "tld", "co")));
+        MatchProcessor processor =
+            new MatchProcessor("_tag", mockSearch, "_name", "domain", "entry", false, true, "domain", maxMatches);
+        IngestDocument ingestDocument =
+            new IngestDocument("_index", "_type", "_id", "_routing", 1L, VersionType.INTERNAL, Map.of("domain", 2));
+
+        // Execute
+        IngestDocument[] holder = new IngestDocument[1];
+        processor.execute(ingestDocument, (result, e) -> holder[0] = result);
+        assertThat(holder[0], notNullValue());
+
+        // Check request
+        SearchRequest request = mockSearch.getCapturedRequest();
+        assertThat(request.source().query(), instanceOf(ConstantScoreQueryBuilder.class));
+        assertThat(((ConstantScoreQueryBuilder) request.source().query()).innerQuery(), instanceOf(TermQueryBuilder.class));
+        TermQueryBuilder termQueryBuilder = (TermQueryBuilder) ((ConstantScoreQueryBuilder) request.source().query()).innerQuery();
+        assertThat(termQueryBuilder.fieldName(), equalTo("domain"));
+        assertThat(termQueryBuilder.value(), equalTo(2));
+
+        // Check result
+        List<?> entries = ingestDocument.getFieldValue("entry", List.class);
+        Map<?, ?> entry = (Map<?, ?>) entries.get(0);
+        assertThat(entry.size(), equalTo(3));
+        assertThat(entry.get("globalRank"), equalTo(451));
+        assertThat(entry.get("tldRank"), equalTo(23));
+        assertThat(entry.get("tld"), equalTo("co"));
+    }
+
+    public void testArray() {
+        int maxMatches = randomIntBetween(1, 8);
+        MockSearchFunction mockSearch =
+            mockedSearchFunction(Map.of(List.of("1", "2"), Map.of("globalRank", 451, "tldRank", 23, "tld", "co")));
+        MatchProcessor processor =
+            new MatchProcessor("_tag", mockSearch, "_name", "domain", "entry", false, true, "domain", maxMatches);
+        IngestDocument ingestDocument =
+            new IngestDocument("_index", "_type", "_id", "_routing", 1L, VersionType.INTERNAL, Map.of("domain", List.of("1", "2")));
+
+        // Execute
+        IngestDocument[] holder = new IngestDocument[1];
+        processor.execute(ingestDocument, (result, e) -> holder[0] = result);
+        assertThat(holder[0], notNullValue());
+
+        // Check request
+        SearchRequest request = mockSearch.getCapturedRequest();
+        assertThat(request.source().query(), instanceOf(ConstantScoreQueryBuilder.class));
+        assertThat(((ConstantScoreQueryBuilder) request.source().query()).innerQuery(), instanceOf(TermsQueryBuilder.class));
+        TermsQueryBuilder termQueryBuilder = (TermsQueryBuilder) ((ConstantScoreQueryBuilder) request.source().query()).innerQuery();
+        assertThat(termQueryBuilder.fieldName(), equalTo("domain"));
+        assertThat(termQueryBuilder.values().size(), equalTo(2));
+        assertThat(termQueryBuilder.values().get(0), equalTo("1"));
+        assertThat(termQueryBuilder.values().get(1), equalTo("2"));
+
+        // Check result
+        List<?> entries = ingestDocument.getFieldValue("entry", List.class);
+        Map<?, ?> entry = (Map<?, ?>) entries.get(0);
+        assertThat(entry.size(), equalTo(3));
+        assertThat(entry.get("globalRank"), equalTo(451));
+        assertThat(entry.get("tldRank"), equalTo(23));
+        assertThat(entry.get("tld"), equalTo("co"));
+    }
+
     private static final class MockSearchFunction implements BiConsumer<SearchRequest, BiConsumer<SearchResponse, Exception>>  {
         private final SearchResponse mockResponse;
         private final SetOnce<SearchRequest> capturedRequest;
@@ -244,13 +308,13 @@ public class MatchProcessorTests extends ESTestCase {
         return new MockSearchFunction(exception);
     }
 
-    public MockSearchFunction mockedSearchFunction(Map<String, Map<String, ?>> documents) {
+    public MockSearchFunction mockedSearchFunction(Map<?, Map<String, ?>> documents) {
         return new MockSearchFunction(mockResponse(documents));
     }
 
-    public SearchResponse mockResponse(Map<String, Map<String, ?>> documents) {
+    public SearchResponse mockResponse(Map<?, Map<String, ?>> documents) {
         SearchHit[] searchHits = documents.entrySet().stream().map(e -> {
-            SearchHit searchHit = new SearchHit(randomInt(100), e.getKey(), Collections.emptyMap());
+            SearchHit searchHit = new SearchHit(randomInt(100), e.getKey().toString(), Collections.emptyMap());
             try (XContentBuilder builder = XContentBuilder.builder(XContentType.SMILE.xContent())) {
                 builder.map(e.getValue());
                 builder.flush();