resolve reverse references for Array types; favor arrayType over scalar fields…

typesystem/src/main/scala/org/apache/hadoop/metadata/query/Expressions.scala

@@ -19,7 +19,7 @@
 package org.apache.hadoop.metadata.query
 
 import org.apache.hadoop.metadata.MetadataException
-import org.apache.hadoop.metadata.types.DataTypes.PrimitiveType
+import org.apache.hadoop.metadata.types.DataTypes.{ArrayType, TypeCategory, PrimitiveType}
 import org.apache.hadoop.metadata.types._
 
 object Expressions {
@@ -367,8 +367,23 @@ object Expressions {
 
   case class FieldExpression(fieldName: String, fieldInfo: FieldInfo, child: Option[Expression])
     extends Expression {
+
+    def elemType(t : IDataType[_]) : IDataType[_] = {
+      if (t.getTypeCategory == TypeCategory.ARRAY ) {
+        val aT = t.asInstanceOf[ArrayType]
+        if ( aT.getElemType.getTypeCategory == TypeCategory.CLASS ||
+          aT.getElemType.getTypeCategory == TypeCategory.STRUCT ) {
+          return aT.getElemType
+        }
+      }
+      t
+    }
+
     val children = if (child.isDefined) List(child.get) else Nil
-    lazy val dataType = if (!fieldInfo.isReverse) fieldInfo.attrInfo.dataType() else fieldInfo.reverseDataType
+    lazy val dataType = {
+      val t = if (!fieldInfo.isReverse) fieldInfo.attrInfo.dataType() else fieldInfo.reverseDataType
+      elemType(t)
+    }
     override lazy val resolved: Boolean = true
 
     override def namedExpressions = if ( child.isDefined ) child.get.namedExpressions else Map()

typesystem/src/main/scala/org/apache/hadoop/metadata/query/TypeUtils.scala

@@ -21,7 +21,7 @@ package org.apache.hadoop.metadata.query
 import java.util.concurrent.atomic.AtomicInteger
 
 import org.apache.hadoop.metadata.MetadataException
-import org.apache.hadoop.metadata.types.DataTypes.PrimitiveType
+import org.apache.hadoop.metadata.types.DataTypes.{ArrayType, TypeCategory, PrimitiveType}
 import org.apache.hadoop.metadata.types._
 
 object TypeUtils {
@@ -72,6 +72,10 @@ object TypeUtils {
     case _ => None
   }
 
+  def hasFields(iDataType: IDataType[_]) : Boolean = {
+    fieldMapping(iDataType).isDefined
+  }
+
   import scala.language.existentials
   case class FieldInfo(dataType : IDataType[_], attrInfo : AttributeInfo, reverseDataType : IDataType[_] = null) {
     def isReverse = reverseDataType != null
@@ -109,11 +113,24 @@ object TypeUtils {
         if (idTypFMap.isDefined) {
           import scala.collection.JavaConversions._
           val fields: Seq[AttributeInfo] = idTypFMap.get.fields.values().filter { aInfo =>
-            aInfo.dataType() == typ
+            aInfo.dataType() == typ ||
+              ( aInfo.dataType().getTypeCategory == TypeCategory.ARRAY &&
+                aInfo.dataType().asInstanceOf[ArrayType].getElemType == typ
+                )
           }.toSeq
           if (fields.size == 1) {
             return Some(FieldInfo(typ, fields(0), idTyp))
           }
+          /*
+           * is there only 1 array field of this type?
+           * If yes resolve to it.
+           * @todo: allow user to specify the relationship to follow by further qualifying the type. for e.g.
+           *   field("LoadProcess.inputTables")
+           */
+          val aFields = fields.filter { aInfo => aInfo.dataType().getTypeCategory == TypeCategory.ARRAY}
+          if (aFields.size == 1) {
+            return Some(FieldInfo(typ, aFields(0), idTyp))
+          }
         }
       } catch {
         case _ : MetadataException => None