expose attribute to type association in HierarchyTypes

typesystem/src/main/java/org/apache/hadoop/metadata/typesystem/types/HierarchicalType.java

@@ -35,6 +35,8 @@ import java.util.Map;
 import java.util.Queue;
 import java.util.Set;
 
+import org.apache.hadoop.metadata.typesystem.types.TypeUtils.Pair;
+
 /**
  * Represents a Type that can have SuperTypes. An Instance of the HierarchicalType can be
  * downcast to a SuperType.
@@ -53,6 +55,7 @@ public abstract class HierarchicalType<ST extends HierarchicalType, T> extends A
     public final int numFields;
     public final ImmutableList<String> superTypes;
     public final ImmutableList<AttributeInfo> immediateAttrs;
+    public final ImmutableMap<String, String> attributeNameToType;
     protected ImmutableMap<String, List<Path>> superTypePaths;
     protected ImmutableMap<String, Path> pathNameToPathMap;
 
@@ -68,6 +71,7 @@ public abstract class HierarchicalType<ST extends HierarchicalType, T> extends A
         this.numFields = numFields;
         this.superTypes = superTypes;
         this.immediateAttrs = null;
+        this.attributeNameToType = null;
     }
 
     HierarchicalType(TypeSystem typeSystem, Class<ST> superTypeClass,
@@ -76,8 +80,10 @@ public abstract class HierarchicalType<ST extends HierarchicalType, T> extends A
         this.typeSystem = typeSystem;
         this.superTypeClass = superTypeClass;
         this.name = name;
-        this.fieldMapping = constructFieldMapping(superTypes,
+        Pair<FieldMapping, ImmutableMap<String, String>> p = constructFieldMapping(superTypes,
                 fields);
+        this.fieldMapping = p.left;
+        this.attributeNameToType = p.right;
         this.numFields = this.fieldMapping.fields.size();
         this.superTypes = superTypes == null ? ImmutableList.<String>of() : superTypes;
         this.immediateAttrs = ImmutableList.<AttributeInfo>copyOf(fields);
@@ -143,13 +149,15 @@ public abstract class HierarchicalType<ST extends HierarchicalType, T> extends A
 
     }
 
-    protected FieldMapping constructFieldMapping(ImmutableList<String> superTypes,
+    protected Pair<FieldMapping, ImmutableMap<String, String>> constructFieldMapping(ImmutableList<String> superTypes,
                                                  AttributeInfo... fields)
     throws MetadataException {
 
         Map<String, AttributeInfo> fieldsMap = new LinkedHashMap<String, AttributeInfo>();
         Map<String, Integer> fieldPos = new HashMap<String, Integer>();
         Map<String, Integer> fieldNullPos = new HashMap<String, Integer>();
+        Map<String, String> attributeNameToType = new HashMap<>();
+
         int numBools = 0;
         int numBytes = 0;
         int numShorts = 0;
@@ -196,6 +204,7 @@ public abstract class HierarchicalType<ST extends HierarchicalType, T> extends A
                 if (fieldsMap.containsKey(attrName)) {
                     attrName = currentPath.addOverrideAttr(attrName);
                 }
+                attributeNameToType.put(attrName, superType.getName());
 
                 fieldsMap.put(attrName, i);
                 fieldNullPos.put(attrName, fieldNullPos.size());
@@ -257,7 +266,7 @@ public abstract class HierarchicalType<ST extends HierarchicalType, T> extends A
         this.superTypePaths = ImmutableMap.copyOf(superTypePaths);
         this.pathNameToPathMap = ImmutableMap.copyOf(pathNameToPathMap);
 
-        return new FieldMapping(fieldsMap,
+        FieldMapping fm =  new FieldMapping(fieldsMap,
                 fieldPos,
                 fieldNullPos,
                 numBools,
@@ -275,6 +284,8 @@ public abstract class HierarchicalType<ST extends HierarchicalType, T> extends A
                 numMaps,
                 numStructs,
                 numReferenceables);
+
+        return new Pair(fm, ImmutableMap.copyOf(attributeNameToType));
     }
 
     public IStruct castAs(IStruct s, String superTypeName) throws MetadataException {
@@ -311,6 +322,22 @@ public abstract class HierarchicalType<ST extends HierarchicalType, T> extends A
         return null;
     }
 
+    public ST getDefinedType(String attrName) throws MetadataException {
+        if (!attributeNameToType.containsKey(attrName)) {
+            throw new MetadataException(String.format("Unknown attribute %s in type %s", attrName, getName()));
+        }
+        return typeSystem.getDataType(superTypeClass, attributeNameToType.get(attrName));
+    }
+
+    public String getDefinedTypeName(String attrName) throws MetadataException {
+        return getDefinedType(attrName).getName();
+    }
+
+    public String getQualifiedName(String attrName) throws MetadataException {
+        String attrTypeName = getDefinedTypeName(attrName);
+        return attrName.contains(".") ? attrName : String.format("%s.%s", attrTypeName, attrName);
+    }
+
     protected Map<String, String> constructDowncastFieldMap(ST subType, Path pathToSubType) {
 
         String pathToSubTypeName = pathToSubType.pathAfterThis;

typesystem/src/main/java/org/apache/hadoop/metadata/typesystem/types/TypeUtils.java

@@ -86,4 +86,14 @@ public class TypeUtils {
         return new TypesDef(JavaConversions.asScalaBuffer(enums), JavaConversions.asScalaBuffer(structs),
                 JavaConversions.asScalaBuffer(traits), JavaConversions.asScalaBuffer(classes));
     }
+
+    protected static class Pair<L,R> {
+        protected L left;
+        protected R right;
+
+        public Pair(L left, R right) {
+            this.left = left;
+            this.right = right;
+        }
+    }
 }

typesystem/src/test/java/org/apache/hadoop/metadata/typesystem/types/TraitTest.java

@@ -27,6 +27,9 @@ import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import static org.apache.hadoop.metadata.typesystem.types.utils.TypesUtil.createOptionalAttrDef;
 import static org.apache.hadoop.metadata.typesystem.types.utils.TypesUtil.createRequiredAttrDef;
 import static org.apache.hadoop.metadata.typesystem.types.utils.TypesUtil.createTraitTypeDef;
@@ -76,6 +79,25 @@ public class TraitTest extends BaseTest {
 
         TraitType DType = (TraitType) getTypeSystem().getDataType(TraitType.class, "D");
 
+//        for(String aName : DType.fieldMapping().fields.keySet()) {
+//            System.out.println(String.format("nameToQualifiedName.put(\"%s\", \"%s\");", aName, DType.getQualifiedName(aName)));
+//        }
+
+        Map<String,String> nameToQualifiedName = new HashMap();
+        {
+            nameToQualifiedName.put("d", "D.d");
+            nameToQualifiedName.put("b", "B.b");
+            nameToQualifiedName.put("c", "C.c");
+            nameToQualifiedName.put("a", "A.a");
+            nameToQualifiedName.put("A.B.D.b", "A.B.D.b");
+            nameToQualifiedName.put("A.B.D.c", "A.B.D.c");
+            nameToQualifiedName.put("A.B.D.d", "A.B.D.d");
+            nameToQualifiedName.put("A.C.D.a", "A.C.D.a");
+            nameToQualifiedName.put("A.C.D.b", "A.C.D.b");
+            nameToQualifiedName.put("A.C.D.c", "A.C.D.c");
+            nameToQualifiedName.put("A.C.D.d", "A.C.D.d");
+        }
+
         Struct s1 = new Struct("D");
         s1.set("d", 1);
         s1.set("c", 1);