Fixes incorrect field in mappings check that ignores sub fields

src/main/kotlin/com/amazon/opendistroforelasticsearch/indexmanagement/rollup/RollupMapperService.kt

@@ -25,6 +25,7 @@ import com.amazon.opendistroforelasticsearch.indexmanagement.rollup.model.dimens
 import com.amazon.opendistroforelasticsearch.indexmanagement.rollup.model.dimension.Histogram
 import com.amazon.opendistroforelasticsearch.indexmanagement.rollup.model.dimension.Terms
 import com.amazon.opendistroforelasticsearch.indexmanagement.rollup.settings.RollupSettings
+import com.amazon.opendistroforelasticsearch.indexmanagement.util.IndexUtils
 import com.amazon.opendistroforelasticsearch.indexmanagement.util.IndexUtils.Companion.PROPERTIES
 import com.amazon.opendistroforelasticsearch.indexmanagement.util.IndexUtils.Companion._META
 import com.amazon.opendistroforelasticsearch.indexmanagement.util._DOC
@@ -197,7 +198,8 @@ class RollupMapperService(
     private fun isFieldInMappings(fieldName: String, mappings: Map<*, *>): Boolean {
         var currMap = mappings
         fieldName.split(".").forEach { field ->
-            val nextMap = (currMap[PROPERTIES] as Map<*, *>?)?.get(field) ?: return false
+            // Looks for a properties field to continue searching or if there is none checks if there is a subfield
+            val nextMap = (currMap[PROPERTIES] as Map<*, *>? ?: currMap[IndexUtils.FIELDS] as Map<*, *>?)?.get(field) ?: return false
             currMap = nextMap as Map<*, *>
         }
 

src/main/kotlin/com/amazon/opendistroforelasticsearch/indexmanagement/util/IndexUtils.kt

@@ -35,6 +35,7 @@ class IndexUtils {
         @Suppress("ObjectPropertyNaming")
         const val _META = "_meta"
         const val PROPERTIES = "properties"
+        const val FIELDS = "fields"
         const val SCHEMA_VERSION = "schema_version"
         const val DEFAULT_SCHEMA_VERSION = 1L
         val logger = LogManager.getLogger(IndexUtils::class.java)

src/test/kotlin/com/amazon/opendistroforelasticsearch/indexmanagement/rollup/RollupMapperServiceTests.kt

@@ -69,6 +69,36 @@ class RollupMapperServiceTests : ESTestCase() {
         }
     }
 
+    fun `test source index validation with subfield`() {
+        val sourceIndex = "test-index"
+
+        val dimensions = listOf(randomDateHistogram().copy(
+            sourceField = "category.keyword"
+        ))
+        val metrics = listOf(randomRollupMetrics().copy(
+            sourceField = "total_quantity"
+        ))
+        val rollup = randomRollup().copy(
+            enabled = true,
+            jobEnabledTime = Instant.now(),
+            dimensions = dimensions,
+            metrics = metrics
+        )
+
+        val client = getClient(getAdminClient(getIndicesAdminClient(
+            getMappingsResponse = getKibanaSampleDataMappingResponse(sourceIndex),
+            getMappingsException = null
+        )))
+        val clusterService = getClusterService()
+        val indexNameExpressionResolver = getIndexNameExpressionResolver(listOf(sourceIndex))
+        val mapperService = RollupMapperService(client, clusterService, indexNameExpressionResolver)
+
+        runBlocking {
+            val sourceIndexValidationResult = mapperService.isSourceIndexValid(rollup)
+            require(sourceIndexValidationResult is RollupJobValidationResult.Valid) { "Source index validation returned unexpected results" }
+        }
+    }
+
     fun `test source index validation with nested field`() {
         val sourceIndex = "test-index"