From 0f5ae59848097fe8b7de1d43094a45edce5cfc3f Mon Sep 17 00:00:00 2001
From: Kaituo Li <kaituo@amazon.com>
Date: Mon, 19 Oct 2020 13:51:02 -0700
Subject: [PATCH] Fix nested field issue
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Previously, our validation logic cannot parse nested fields from the getFieldMapping response because nested fields don't have the full name in the response.  For example, the field nested_host.host only has "host" in the type mapping instead of "nested_host.host".  This PR fixes that by not depending on field name when parsing the name type mapping.
Testing done:
1. manual testing passes.
---
 .../IndexAnomalyDetectorActionHandler.java    | 28 +++++++++++++------
 1 file changed, 20 insertions(+), 8 deletions(-)

diff --git a/src/main/java/com/amazon/opendistroforelasticsearch/ad/rest/handler/IndexAnomalyDetectorActionHandler.java b/src/main/java/com/amazon/opendistroforelasticsearch/ad/rest/handler/IndexAnomalyDetectorActionHandler.java
index da30b799..f1799495 100644
--- a/src/main/java/com/amazon/opendistroforelasticsearch/ad/rest/handler/IndexAnomalyDetectorActionHandler.java
+++ b/src/main/java/com/amazon/opendistroforelasticsearch/ad/rest/handler/IndexAnomalyDetectorActionHandler.java
@@ -287,25 +287,37 @@ private void validateCategoricalField(String detectorId) {
             // example getMappingsResponse:
             // GetFieldMappingsResponse{mappings={server-metrics={_doc={service=FieldMappingMetadata{fullName='service',
             // source=org.elasticsearch.common.bytes.BytesArray@7ba87dbd}}}}}
+            // for nested field, it would be
+            // GetFieldMappingsResponse{mappings={server-metrics={_doc={host_nest.host2=FieldMappingMetadata{fullName='host_nest.host2',
+            // source=org.elasticsearch.common.bytes.BytesArray@8fb4de08}}}}}
             boolean foundField = false;
             Map<String, Map<String, Map<String, FieldMappingMetadata>>> mappingsByIndex = getMappingsResponse.mappings();
 
             for (Map<String, Map<String, FieldMappingMetadata>> mappingsByType : mappingsByIndex.values()) {
                 for (Map<String, FieldMappingMetadata> mappingsByField : mappingsByType.values()) {
                     for (Map.Entry<String, FieldMappingMetadata> field2Metadata : mappingsByField.entrySet()) {
+                        // example output:
+                        // host_nest.host2=FieldMappingMetadata{fullName='host_nest.host2',
+                        // source=org.elasticsearch.common.bytes.BytesArray@8fb4de08}
                         FieldMappingMetadata fieldMetadata = field2Metadata.getValue();
 
                         if (fieldMetadata != null) {
-                            Object metadata = fieldMetadata.sourceAsMap().get(categoryField0);
-                            if (metadata != null && metadata instanceof Map) {
-                                foundField = true;
-                                Map<String, Object> metadataMap = (Map<String, Object>) metadata;
-                                String typeName = (String) metadataMap.get(CommonName.TYPE);
-                                if (!typeName.equals(CommonName.KEYWORD_TYPE) && !typeName.equals(CommonName.IP_TYPE)) {
-                                    listener.onFailure(new IllegalArgumentException(CATEGORICAL_FIELD_TYPE_ERR_MSG));
-                                    return;
+                            // sourceAsMap returns sth like {host2={type=keyword}} with host2 being a nested field
+                            Map<String, Object> fieldMap = fieldMetadata.sourceAsMap();
+                            if (fieldMap != null) {
+                                for (Object type : fieldMap.values()) {
+                                    if (type != null && type instanceof Map) {
+                                        foundField = true;
+                                        Map<String, Object> metadataMap = (Map<String, Object>) type;
+                                        String typeName = (String) metadataMap.get(CommonName.TYPE);
+                                        if (!typeName.equals(CommonName.KEYWORD_TYPE) && !typeName.equals(CommonName.IP_TYPE)) {
+                                            listener.onFailure(new IllegalArgumentException(CATEGORICAL_FIELD_TYPE_ERR_MSG));
+                                            return;
+                                        }
+                                    }
                                 }
                             }
+
                         }
                     }
                 }