introduce ClassType

src/main/java/org/apache/metadata/IReferencableInstance.java → src/main/java/org/apache/metadata/IReferenceableInstance.java

@@ -25,7 +25,7 @@ import org.apache.metadata.storage.Id;
  * Represents and instance of a ClassType. These have identity.
  * Transient instances will have a UNASSIGNED identity.
  */
-public interface IReferencableInstance extends IStruct {
+public interface IReferenceableInstance extends IStruct {
 
     ImmutableList<String> getTraits();
     Id getId();

src/main/java/org/apache/metadata/ITypedReferencableInstance.java → src/main/java/org/apache/metadata/ITypedReferenceableInstance.java

@@ -18,5 +18,5 @@
 
 package org.apache.metadata;
 
-public interface ITypedReferencableInstance extends ITypedInstance, IReferencableInstance {
+public interface ITypedReferenceableInstance extends ITypedInstance, IReferenceableInstance {
 }

src/main/java/org/apache/metadata/Referenceable.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.metadata;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.apache.metadata.storage.Id;
+
+import java.util.Map;
+
+/**
+ * Represents a Class Instance that has not been associated with a FieldMapping.
+ */
+public class Referenceable extends Struct implements IReferenceableInstance {
+
+    private final Id id;
+    private final ImmutableMap<String, IStruct> traits;
+    private final ImmutableList<String> traitNames;
+
+    public Referenceable(String typeName, String... traitNames) {
+        super(typeName);
+        id = new Id(typeName);
+        this.traitNames = ImmutableList.copyOf(traitNames);
+        ImmutableMap.Builder<String,IStruct> b = new ImmutableMap.Builder<String, IStruct>();
+        for(String t : traitNames) {
+            b.put(t, new Struct(t));
+        }
+        traits = b.build();
+    }
+
+    public Referenceable(String typeName, Map<String, Object> values) {
+        super(typeName, values);
+        id = new Id(typeName);
+        traitNames = ImmutableList.of();
+        traits = ImmutableMap.of();
+    }
+
+    @Override
+    public ImmutableList<String> getTraits() {
+        return traitNames;
+    }
+
+    @Override
+    public Id getId() {
+        return id;
+    }
+
+    @Override
+    public IStruct getTrait(String typeName) {
+        return traits.get(typeName);
+    }
+}

src/main/java/org/apache/metadata/storage/Id.java

@@ -20,23 +20,25 @@ package org.apache.metadata.storage;
 
 public class Id {
 
-    public static final int UNASSIGNED = -1;
-
-    public final int id;
+    public final long id;
     public final String className;
     public final int version;
 
-    public Id(int id, int version, String className) {
+    public Id(long id, int version, String className) {
         this.id = id;
         this.className = className;
         this.version = version;
     }
 
     public Id(String className) {
-        this(UNASSIGNED, 0, className);
+        this(- System.currentTimeMillis(), 0, className);
     }
 
     public boolean isUnassigned() {
-        return id == UNASSIGNED;
+        return id < 0;
+    }
+
+    public String toString() {
+        return String.format("(type: %s, id: %s)", className, isUnassigned() ? "<unassigned>" : "" + id);
     }
 }

src/main/java/org/apache/metadata/storage/ReferenceableInstance.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.metadata.storage;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.apache.metadata.IStruct;
+import org.apache.metadata.ITypedReferenceableInstance;
+import org.apache.metadata.ITypedStruct;
+import org.apache.metadata.types.FieldMapping;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.Date;
+
+/*
+ * @todo handle names prefixed by traitName.
+ */
+public class ReferenceableInstance extends StructInstance implements ITypedReferenceableInstance {
+
+    private final Id id;
+    private final ImmutableMap<String, ITypedStruct> traits;
+    private final ImmutableList<String> traitNames;
+
+
+    public ReferenceableInstance(Id id, String dataTypeName, FieldMapping fieldMapping, boolean[] nullFlags,
+                                 boolean[] bools, byte[] bytes, short[] shorts, int[] ints, long[] longs,
+                                 float[] floats, double[] doubles, BigDecimal[] bigDecimals,
+                                 BigInteger[] bigIntegers, Date[] dates, String[] strings,
+                                 ImmutableList<Object>[] arrays, ImmutableMap<Object, Object>[] maps,
+                                 StructInstance[] structs,
+                                 ReferenceableInstance[] referenceableInstances,
+                                 Id[] ids,
+                                 ImmutableMap<String, ITypedStruct> traits) {
+        super(dataTypeName, fieldMapping, nullFlags, bools, bytes, shorts, ints, longs, floats, doubles, bigDecimals,
+                bigIntegers, dates, strings, arrays, maps, structs, referenceableInstances, ids);
+        this.id = id;
+        this.traits = traits;
+        ImmutableList.Builder<String> b = new ImmutableList.Builder<String>();
+        for(String t : traits.keySet()) {
+            b.add(t);
+        }
+        this.traitNames = b.build();
+    }
+
+    @Override
+    public ImmutableList<String> getTraits() {
+        return traitNames;
+    }
+
+    @Override
+    public Id getId() {
+        return id;
+    }
+
+    @Override
+    public IStruct getTrait(String typeName) {
+        return traits.get(typeName);
+    }
+}

src/main/java/org/apache/metadata/storage/StructInstance.java

@@ -49,12 +49,15 @@ public class StructInstance implements ITypedStruct {
     public final ImmutableList<Object>[] arrays;
     public final ImmutableMap<Object,Object>[] maps;
     public final StructInstance[] structs;
+    public final ReferenceableInstance[] referenceables;
+    public final Id[] ids;
 
     public StructInstance(String dataTypeName, FieldMapping fieldMapping,
                           boolean[] nullFlags, boolean[] bools, byte[] bytes, short[] shorts, int[] ints,
                           long[] longs, float[] floats, double[] doubles,
                           BigDecimal[] bigDecimals, BigInteger[] bigIntegers, Date[] dates, String[] strings,
-                          ImmutableList<Object>[] arrays, ImmutableMap<Object, Object>[] maps, StructInstance[] structs) {
+                          ImmutableList<Object>[] arrays, ImmutableMap<Object, Object>[] maps,
+                          StructInstance[] structs, ReferenceableInstance[] referenceables, Id[] ids) {
         assert dataTypeName != null;
         this.dataTypeName = dataTypeName;
         this.fieldMapping = fieldMapping;
@@ -73,6 +76,8 @@ public class StructInstance implements ITypedStruct {
         this.arrays = arrays;
         this.maps = maps;
         this.structs = structs;
+        this.referenceables = referenceables;
+        this.ids = ids;
 
         for(int i=0; i<nullFlags.length; i++) {
             nullFlags[i] = true;

src/main/java/org/apache/metadata/types/AttributeInfo.java

@@ -33,8 +33,9 @@ public class AttributeInfo {
     public final String reverseAttributeName;
 
     AttributeInfo(TypeSystem t, AttributeDefinition def) throws MetadataException {
+        TypeUtils.validateName(def.name);
         this.name = def.name;
-        this.dataType = t.getDataType(def.dataTypeName);
+        this.dataType = t.getDataType(IDataType.class, def.dataTypeName);
         this.multiplicity = def.multiplicity;
         this.isComposite = def.isComposite;
         this.reverseAttributeName = def.reverseAttributeName;

src/main/java/org/apache/metadata/types/ClassType.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.metadata.types;
+
+import com.google.common.collect.ImmutableBiMap;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.apache.metadata.*;
+import org.apache.metadata.storage.Id;
+import org.apache.metadata.storage.ReferenceableInstance;
+import org.apache.metadata.storage.StructInstance;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.Date;
+import java.util.Map;
+
+public class ClassType extends HierarchicalType<ClassType, IReferenceableInstance>
+        implements IConstructableType<IReferenceableInstance, ITypedReferenceableInstance> {
+
+    public static final String TRAIT_NAME_SEP = "::";
+
+    /**
+     * Used when creating a ClassType, to support recursive Structs.
+     */
+    ClassType(TypeSystem typeSystem, String name, ImmutableList<String> superTypes, int numFields) {
+        super(typeSystem, name, superTypes, numFields);
+    }
+
+    ClassType(TypeSystem typeSystem, String name, ImmutableList<String> superTraits, AttributeInfo... fields)
+            throws MetadataException {
+        super(typeSystem, name, superTraits, fields);
+    }
+
+    @Override
+    public DataTypes.TypeCategory getTypeCategory() {
+        return DataTypes.TypeCategory.CLASS;
+    }
+
+    public void validateId(Id id) throws MetadataException {
+        if ( id != null ) {
+            ClassType cType = typeSystem.getDataType(ClassType.class, id.className);
+            if ( isSubType(cType.getName()) ) {
+                return;
+            }
+            throw new MetadataException(String.format("Id %s is not valid for class %s", id, getName()));
+        }
+    }
+
+    @Override
+    public ITypedReferenceableInstance convert(Object val, Multiplicity m) throws MetadataException {
+
+        if ( val != null ) {
+            if ( val instanceof Struct) {
+                Struct s = (Struct) val;
+                Referenceable r = null;
+
+                if ( s.typeName != getName() ) {
+                    throw new ValueConversionException(this, val);
+                }
+
+                if ( val instanceof Referenceable ) {
+                     r = (Referenceable)val;
+                }
+
+                ITypedReferenceableInstance tr = r != null ?
+                        createInstanceWithTraits(r, r.getTraits().toArray(new String[0])) : createInstance();
+
+                for(Map.Entry<String,AttributeInfo> e : fieldMapping.fields.entrySet() ) {
+                    String attrKey = e.getKey();
+                    AttributeInfo i = e.getValue();
+                    Object aVal = s.get(attrKey);
+                    try {
+                        tr.set(attrKey, aVal);
+                    } catch(ValueConversionException ve) {
+                        throw new ValueConversionException(this, val, ve);
+                    }
+                }
+
+                return tr;
+            } else if ( val instanceof ReferenceableInstance ) {
+                validateId(((ReferenceableInstance)val).getId());
+                return (ReferenceableInstance) val;
+            } else {
+                throw new ValueConversionException(this, val);
+            }
+        }
+        if (!m.nullAllowed() ) {
+            throw new ValueConversionException.NullConversionException(m);
+        }
+        return null;
+    }
+
+    @Override
+    public ITypedReferenceableInstance createInstance() throws MetadataException {
+        return createInstanceWithTraits(null);
+    }
+
+    public ITypedReferenceableInstance createInstanceWithTraits(Referenceable r, String... traitNames)
+    throws MetadataException {
+
+        ImmutableMap.Builder<String, ITypedStruct> b = new ImmutableBiMap.Builder<String, ITypedStruct>();
+        for(String t : traitNames) {
+            TraitType tType = typeSystem.getDataType(TraitType.class, t);
+            IStruct iTraitObject = r == null ? null : r.getTrait(t);
+            ITypedStruct trait = iTraitObject == null ? tType.createInstance() :
+                    tType.convert(iTraitObject, Multiplicity.REQUIRED);
+            b.put(t, trait);
+        }
+
+        return new ReferenceableInstance(new Id(getName()),
+                getName(),
+                fieldMapping,
+                new boolean[fieldMapping.fields.size()],
+                fieldMapping.numBools == 0 ? null : new boolean[fieldMapping.numBools],
+                fieldMapping.numBytes == 0 ? null : new byte[fieldMapping.numBytes],
+                fieldMapping.numShorts == 0 ? null : new short[fieldMapping.numShorts],
+                fieldMapping.numInts == 0 ? null : new int[fieldMapping.numInts],
+                fieldMapping.numLongs == 0 ? null : new long[fieldMapping.numLongs],
+                fieldMapping.numFloats == 0 ? null : new float[fieldMapping.numFloats],
+                fieldMapping.numDoubles == 0 ? null : new double[fieldMapping.numDoubles],
+                fieldMapping.numBigDecimals == 0 ? null : new BigDecimal[fieldMapping.numBigDecimals],
+                fieldMapping.numBigInts == 0 ? null : new BigInteger[fieldMapping.numBigInts],
+                fieldMapping.numDates == 0 ? null : new Date[fieldMapping.numDates],
+                fieldMapping.numStrings == 0 ? null : new String[fieldMapping.numStrings],
+                fieldMapping.numArrays == 0 ? null : new ImmutableList[fieldMapping.numArrays],
+                fieldMapping.numMaps == 0 ? null : new ImmutableMap[fieldMapping.numMaps],
+                fieldMapping.numStructs == 0 ? null : new StructInstance[fieldMapping.numStructs],
+                fieldMapping.numReferenceables == 0 ? null : new ReferenceableInstance[fieldMapping.numReferenceables],
+                fieldMapping.numReferenceables == 0 ? null : new Id[fieldMapping.numReferenceables],
+                b.build());
+    }
+
+    @Override
+    public void output(IReferenceableInstance s, Appendable buf, String prefix) throws MetadataException {
+        TypeUtils.outputVal("{", buf, prefix);
+        if ( s == null ) {
+            TypeUtils.outputVal("<null>\n", buf, "");
+            return;
+        }
+        TypeUtils.outputVal("\n", buf, "");
+        String fieldPrefix = prefix + "\t";
+
+        TypeUtils.outputVal("id : ", buf, fieldPrefix);
+        TypeUtils.outputVal(s.getId().toString(), buf, "");
+        TypeUtils.outputVal("\n", buf, "");
+
+        for(AttributeInfo i : fieldMapping.fields.values()) {
+            Object aVal = s.get(i.name);
+            TypeUtils.outputVal(i.name + " : ", buf, fieldPrefix);
+            i.dataType().output(aVal, buf, "");
+            TypeUtils.outputVal("\n", buf, "");
+        }
+
+        for(String sT : s.getTraits() ) {
+            TraitType tt = typeSystem.getDataType(TraitType.class, sT);
+            TypeUtils.outputVal(sT + " : ", buf, fieldPrefix);
+            tt.output(s.getTrait(sT), buf, fieldPrefix);
+        }
+
+        TypeUtils.outputVal("}", buf, fieldPrefix);
+    }
+
+}
\ No newline at end of file

src/main/java/org/apache/metadata/types/ITypeBrowser.java → src/main/java/org/apache/metadata/types/ClassTypeDefinition.java

@@ -18,10 +18,18 @@
 
 package org.apache.metadata.types;
 
-/**
- * @nopublic
- */
-public interface ITypeBrowser {
+import com.google.common.collect.ImmutableList;
+
+public class ClassTypeDefinition {
+
+    public final String typeName;
+    public final ImmutableList<String> superTraits;
+    public final AttributeDefinition[] attributeDefinitions;
 
-    IDataType dataType(String name);
-}
+    public ClassTypeDefinition(String typeName, ImmutableList<String> superTraits,
+                               AttributeDefinition[] attributeDefinitions) {
+        this.typeName = typeName;
+        this.superTraits = superTraits == null ? ImmutableList.<String>of() : superTraits;
+        this.attributeDefinitions = attributeDefinitions;
+    }
+}
\ No newline at end of file

src/main/java/org/apache/metadata/types/DataTypes.java

@@ -382,7 +382,7 @@ public class DataTypes {
     static String ARRAY_TYPE_SUFFIX = ">";
     public static class ArrayType extends AbstractDataType<ImmutableCollection<?>> {
 
-        private final IDataType elemType;
+        private IDataType elemType;
         private final String nm;
 
         public ArrayType(IDataType elemType) {
@@ -395,6 +395,10 @@ public class DataTypes {
             return elemType;
         }
 
+        protected void setElemType(IDataType elemType) {
+            this.elemType = elemType;
+        }
+
         @Override
         public String getName() {
             return nm;
@@ -445,8 +449,8 @@ public class DataTypes {
     static String MAP_TYPE_SUFFIX = ">";
     public static class MapType extends AbstractDataType<ImmutableMap<?, ?>> {
 
-        private final IDataType keyType;
-        private final IDataType valueType;
+        private IDataType keyType;
+        private IDataType valueType;
         private final String nm;
 
         public MapType(IDataType keyType, IDataType valueType) {
@@ -466,6 +470,14 @@ public class DataTypes {
             return valueType;
         }
 
+        protected void setKeyType(IDataType keyType) {
+            this.keyType = keyType;
+        }
+
+        protected void setValueType(IDataType valueType) {
+            this.keyType = valueType;
+        }
+
         @Override
         public String getName() {
             return nm;

src/main/java/org/apache/metadata/types/FieldMapping.java

@@ -20,6 +20,9 @@ package org.apache.metadata.types;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import org.apache.metadata.MetadataException;
+import org.apache.metadata.MetadataService;
+import org.apache.metadata.storage.Id;
+import org.apache.metadata.storage.ReferenceableInstance;
 import org.apache.metadata.storage.StructInstance;
 
 import java.math.BigDecimal;
@@ -48,11 +51,13 @@ public class FieldMapping {
     public final int numArrays;
     public final int numMaps;
     public final int numStructs;
+    public final int numReferenceables;
 
     public FieldMapping(Map<String, AttributeInfo> fields, Map<String, Integer> fieldPos,
                         Map<String, Integer> fieldNullPos, int numBools, int numBytes, int numShorts,
                         int numInts, int numLongs, int numFloats, int numDoubles, int numBigInts, int numBigDecimals,
-                        int numDates, int numStrings, int numArrays, int numMaps, int numStructs) {
+                        int numDates, int numStrings, int numArrays, int numMaps, int numStructs,
+                        int numReferenceables) {
         this.fields = fields;
         this.fieldPos = fieldPos;
         this.fieldNullPos = fieldNullPos;
@@ -70,6 +75,7 @@ public class FieldMapping {
         this.numArrays = numArrays;
         this.numMaps = numMaps;
         this.numStructs = numStructs;
+        this.numReferenceables = numReferenceables;
     }
 
     public void set(StructInstance s, String attrName, Object val) throws MetadataException {
@@ -79,7 +85,16 @@ public class FieldMapping {
         }
         int pos = fieldPos.get(attrName);
         int nullPos = fieldNullPos.get(attrName);
-        Object cVal = i.dataType().convert(val, i.multiplicity);
+        Object cVal = null;
+
+        if (val != null && val instanceof Id) {
+            ClassType clsType =
+                    MetadataService.getCurrentTypeSystem().getDataType(ClassType.class, i.dataType().getName());
+            clsType.validateId((Id)cVal);
+            cVal = val;
+        } else {
+            cVal = i.dataType().convert(val, i.multiplicity);
+        }
         if ( cVal == null ) {
             s.nullFlags[nullPos] = true;
             return;
@@ -111,8 +126,15 @@ public class FieldMapping {
             s.arrays[pos] = (ImmutableList) cVal;
         } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.MAP ) {
             s.maps[pos] = (ImmutableMap) cVal;
-        } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.STRUCT ) {
+        } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.STRUCT ||
+                i.dataType().getTypeCategory() == DataTypes.TypeCategory.TRAIT ) {
             s.structs[pos] = (StructInstance) cVal;
+        } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.CLASS ) {
+            if ( cVal instanceof  Id ) {
+                s.ids[pos] = (Id) cVal;
+            } else {
+                s.referenceables[pos] = (ReferenceableInstance) cVal;
+            }
         } else {
             throw new MetadataException(String.format("Unknown datatype %s", i.dataType()));
         }

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

@@ -35,7 +35,7 @@ import java.util.*;
 public abstract class HierarchicalType<ST extends HierarchicalType,T> extends AbstractDataType<T>
         implements Comparable<ST> {
 
-    public final ITypeBrowser typeSystem;
+    public final TypeSystem typeSystem;
     public final String name;
     public final FieldMapping fieldMapping;
     public final int numFields;
@@ -47,7 +47,7 @@ public abstract class HierarchicalType<ST extends HierarchicalType,T> extends Ab
     /**
      * Used when creating a Type, to support recursive Structs.
      */
-    HierarchicalType(ITypeBrowser typeSystem, String name, ImmutableList<String> superTypes, int numFields) {
+    HierarchicalType(TypeSystem typeSystem, String name, ImmutableList<String> superTypes, int numFields) {
         this.typeSystem = typeSystem;
         this.name = name;
         this.fieldMapping = null;
@@ -56,7 +56,7 @@ public abstract class HierarchicalType<ST extends HierarchicalType,T> extends Ab
         this.immediateAttrs = null;
     }
 
-    HierarchicalType(ITypeBrowser typeSystem, String name, ImmutableList<String> superTypes, AttributeInfo... fields)
+    HierarchicalType(TypeSystem typeSystem, String name, ImmutableList<String> superTypes, AttributeInfo... fields)
             throws MetadataException {
         this.typeSystem = typeSystem;
         this.name = name;
@@ -67,6 +67,24 @@ public abstract class HierarchicalType<ST extends HierarchicalType,T> extends Ab
         this.immediateAttrs = ImmutableList.<AttributeInfo>copyOf(fields);
     }
 
+    @Override
+    public String getName() {
+        return name;
+    }
+
+    public FieldMapping fieldMapping() {
+        return fieldMapping;
+    }
+
+    /**
+     * Given type must be a SubType of this type.
+     * @param typeName
+     * @throws MetadataException
+     */
+    public boolean isSubType(String typeName) throws MetadataException {
+        HierarchicalType cType = typeSystem.getDataType(HierarchicalType.class, typeName);
+        return  ( cType == this || cType.superTypePaths.containsKey(getName()) );
+    }
 
     protected FieldMapping constructFieldMapping(ImmutableList<String> superTraits,
                                                  AttributeInfo... fields)
@@ -89,6 +107,7 @@ public abstract class HierarchicalType<ST extends HierarchicalType,T> extends Ab
         int numArrays = 0;
         int numMaps = 0;
         int numStructs = 0;
+        int numReferenceables = 0;
 
         Map<String, List<Path>> superTypePaths = new HashMap<String, List<Path>>();
         Map<String, Path> pathNameToPathMap = new HashMap<String, Path>();
@@ -167,9 +186,13 @@ public abstract class HierarchicalType<ST extends HierarchicalType,T> extends Ab
                 } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.MAP ) {
                     fieldPos.put(attrName, numMaps);
                     numMaps++;
-                } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.STRUCT ) {
+                } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.STRUCT ||
+                        i.dataType().getTypeCategory() == DataTypes.TypeCategory.TRAIT ) {
                     fieldPos.put(attrName, numStructs);
                     numStructs++;
+                }  else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.CLASS ) {
+                    fieldPos.put(attrName, numReferenceables);
+                    numReferenceables++;
                 } else {
                     throw new MetadataException(String.format("Unknown datatype %s", i.dataType()));
                 }
@@ -201,7 +224,8 @@ public abstract class HierarchicalType<ST extends HierarchicalType,T> extends Ab
                 numStrings,
                 numArrays,
                 numMaps,
-                numStructs);
+                numStructs,
+                numReferenceables);
     }
 
     public IStruct castAs(IStruct s, String superTypeName) throws MetadataException {

src/main/java/org/apache/metadata/types/IConstructableType.java

@@ -2,9 +2,10 @@ package org.apache.metadata.types;
 
 
 import org.apache.metadata.ITypedInstance;
+import org.apache.metadata.MetadataException;
 
 public interface IConstructableType<U, T extends ITypedInstance> extends IDataType<U> {
 
-    T createInstance();
+    T createInstance() throws MetadataException;
     FieldMapping fieldMapping();
 }

src/main/java/org/apache/metadata/types/StructType.java

@@ -32,7 +32,7 @@ import java.util.*;
 public class StructType  extends AbstractDataType<IStruct>
         implements IConstructableType<IStruct, ITypedStruct> {
 
-    public final ITypeBrowser typeSystem;
+    public final TypeSystem typeSystem;
     public final String name;
     public final FieldMapping fieldMapping;
     public final int numFields;
@@ -41,7 +41,7 @@ public class StructType  extends AbstractDataType<IStruct>
     /**
      * Used when creating a StructType, to support recursive Structs.
      */
-    protected StructType(ITypeBrowser typeSystem, String name, int numFields) {
+    protected StructType(TypeSystem typeSystem, String name, int numFields) {
         this.typeSystem = typeSystem;
         this.name = name;
         this.fieldMapping = null;
@@ -49,7 +49,7 @@ public class StructType  extends AbstractDataType<IStruct>
         this.handler = null;
     }
 
-    protected StructType(ITypeBrowser typeSystem, String name,
+    protected StructType(TypeSystem typeSystem, String name,
                          ImmutableList<String> superTypes, AttributeInfo... fields) throws MetadataException {
         this.typeSystem = typeSystem;
         this.name = name;
@@ -89,6 +89,7 @@ public class StructType  extends AbstractDataType<IStruct>
         int numArrays = 0;
         int numMaps = 0;
         int numStructs = 0;
+        int numReferenceables = 0;
 
         for(AttributeInfo i : fields) {
             if ( fieldsMap.containsKey(i.name) ) {
@@ -136,10 +137,14 @@ public class StructType  extends AbstractDataType<IStruct>
             } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.MAP ) {
                 fieldPos.put(i.name, numMaps);
                 numMaps++;
-            } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.STRUCT ) {
+            } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.STRUCT  ||
+                    i.dataType().getTypeCategory() == DataTypes.TypeCategory.TRAIT ) {
                 fieldPos.put(i.name, numStructs);
                 numStructs++;
-            } else {
+            } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.CLASS ) {
+                fieldPos.put(i.name, numReferenceables);
+                numReferenceables++;
+            }  else {
                 throw new MetadataException(String.format("Unknown datatype %s", i.dataType()));
             }
         }
@@ -160,7 +165,8 @@ public class StructType  extends AbstractDataType<IStruct>
                 numStrings,
                 numArrays,
                 numMaps,
-                numStructs);
+                numStructs,
+                numReferenceables);
     }
 
 

src/main/java/org/apache/metadata/types/TraitType.java

@@ -31,27 +31,18 @@ public class TraitType extends HierarchicalType<TraitType, IStruct>
     /**
      * Used when creating a TraitType, to support recursive Structs.
      */
-    TraitType(ITypeBrowser typeSystem, String name, ImmutableList<String> superTraits, int numFields) {
+    TraitType(TypeSystem typeSystem, String name, ImmutableList<String> superTraits, int numFields) {
         super(typeSystem, name, superTraits, numFields);
         handler = null;
     }
 
-    TraitType(ITypeBrowser typeSystem, String name, ImmutableList<String> superTraits, AttributeInfo... fields)
+    TraitType(TypeSystem typeSystem, String name, ImmutableList<String> superTraits, AttributeInfo... fields)
             throws MetadataException {
         super(typeSystem, name, superTraits, fields);
         handler = new TypedStructHandler(this);
     }
 
     @Override
-    public String getName() {
-        return name;
-    }
-
-    public FieldMapping fieldMapping() {
-        return fieldMapping;
-    }
-
-    @Override
     public DataTypes.TypeCategory getTypeCategory() {
         return DataTypes.TypeCategory.TRAIT;
     }

src/main/java/org/apache/metadata/types/TypeSystem.java

@@ -23,7 +23,7 @@ import org.apache.metadata.MetadataException;
 
 import java.util.*;
 
-public class TypeSystem implements ITypeBrowser {
+public class TypeSystem {
 
     private Map<String, IDataType> types;
 
@@ -32,6 +32,8 @@ public class TypeSystem implements ITypeBrowser {
         registerPrimitiveTypes();
     }
 
+    private TypeSystem(TypeSystem ts) {}
+
     public ImmutableList<String> getTypeNames() {
         return ImmutableList.copyOf(types.keySet());
     }
@@ -54,10 +56,30 @@ public class TypeSystem implements ITypeBrowser {
         return types.get(name);
     }
 
-    public IDataType getDataType(String name) throws MetadataException {
+    public <T> T getDataType(Class<T> cls, String name) throws MetadataException {
         if ( types.containsKey(name) ) {
-            return types.get(name);
+            return cls.cast(types.get(name));
+        }
+
+        /*
+         * is this an Array Type?
+         */
+        String arrElemType = TypeUtils.parseAsArrayType(name);
+        if ( arrElemType != null ) {
+            IDataType dT = defineArrayType(getDataType(IDataType.class, arrElemType));
+            return cls.cast(dT);
+        }
+
+        /*
+         * is this a Map Type?
+         */
+        String[] mapType = TypeUtils.parseAsMapType(name);
+        if ( mapType != null ) {
+            IDataType dT = defineMapType(getDataType(IDataType.class, mapType[0]),
+                    getDataType(IDataType.class, mapType[1]));
+            return cls.cast(dT);
         }
+
         throw new MetadataException(String.format("Unknown datatype: %s", name));
     }
 
@@ -160,6 +182,9 @@ public class TypeSystem implements ITypeBrowser {
         List<TraitType> l = new ArrayList<TraitType>(transientTypes.traitTypes.values());
         Collections.sort(l);
         List<AttributeInfo> recursiveRefs = new ArrayList<AttributeInfo>();
+        List<DataTypes.ArrayType> recursiveArrayTypes = new ArrayList<DataTypes.ArrayType>();
+        List<DataTypes.MapType> recursiveMapTypes = new ArrayList<DataTypes.MapType>();
+
 
         try {
             for (TraitType ttO : l) {
@@ -170,20 +195,40 @@ public class TypeSystem implements ITypeBrowser {
                     if (transientTypes.traitTypes.containsKey(traitDef.attributeDefinitions[i].dataTypeName)) {
                         recursiveRefs.add(infos[i]);
                     }
+                    if ( infos[i].dataType().getTypeCategory() == DataTypes.TypeCategory.ARRAY ) {
+                        DataTypes.ArrayType arrType = (DataTypes.ArrayType) infos[i].dataType();
+                        if (transientTypes.traitTypes.containsKey(arrType.getElemType().getName())) {
+                            recursiveArrayTypes.add(arrType);
+                        }
+                    }
+                    if ( infos[i].dataType().getTypeCategory() == DataTypes.TypeCategory.MAP ) {
+                        DataTypes.MapType mapType = (DataTypes.MapType) infos[i].dataType();
+                        if (transientTypes.traitTypes.containsKey(mapType.getKeyType().getName())) {
+                            recursiveMapTypes.add(mapType);
+                        } else if (transientTypes.traitTypes.containsKey(mapType.getValueType().getName())) {
+                            recursiveMapTypes.add(mapType);
+                        }
+                    }
                 }
 
                 TraitType tt = new TraitType(this, traitDef.typeName, traitDef.superTraits, infos);
-                ;
                 types.put(tt.getName(), tt);
             }
 
             /*
              * Step 4:
-             * - fix up references in recursive AttrInfo
+             * - fix up references in recursive AttrInfo and recursive Collection Types.
              */
             for (AttributeInfo info : recursiveRefs) {
                 info.setDataType(dataType(info.dataType().getName()));
             }
+            for(DataTypes.ArrayType arrType : recursiveArrayTypes ) {
+                arrType.setElemType(dataType(arrType.getElemType().getName()));
+            }
+            for(DataTypes.MapType mapType : recursiveMapTypes ) {
+                mapType.setKeyType(dataType(mapType.getKeyType().getName()));
+                mapType.setValueType(dataType(mapType.getValueType().getName()));
+            }
         } catch(MetadataException me) {
             for(String sT : transientTypes.traitTypes.keySet()) {
                 types.remove(sT);
@@ -211,13 +256,54 @@ public class TypeSystem implements ITypeBrowser {
         return dT;
     }
 
-    class TransientTypeSystem implements ITypeBrowser {
+    class TransientTypeSystem extends TypeSystem {
         Map<String, TraitType> traitTypes = new HashMap<String, TraitType>();
 
+        TransientTypeSystem() {
+            super(TypeSystem.this);
+        }
+
         public IDataType dataType(String name) {
             IDataType dT = TypeSystem.this.dataType(name);
             dT = dT == null ? traitTypes.get(name) : dT;
             return dT;
         }
+
+        @Override
+        public ImmutableList<String> getTypeNames() {
+            return TypeSystem.this.getTypeNames();
+        }
+
+        @Override
+        public <T> T getDataType(Class<T> cls, String name) throws MetadataException  {
+            return TypeSystem.this.getDataType(cls, name);
+        }
+
+        @Override
+        public StructType defineStructType(String name, boolean errorIfExists, AttributeDefinition... attrDefs)
+                throws MetadataException {
+            return TypeSystem.this.defineStructType(name, errorIfExists, attrDefs);
+        }
+
+        @Override
+        public TraitType defineTraitType(boolean errorIfExists, TraitTypeDefinition traitDef) throws MetadataException {
+            return TypeSystem.this.defineTraitType(errorIfExists, traitDef);
+        }
+
+        @Override
+        public Map<String, TraitType> defineTraitTypes(boolean errorIfExists, TraitTypeDefinition... traitDefs)
+                throws MetadataException {
+            return TypeSystem.this.defineTraitTypes(errorIfExists, traitDefs);
+        }
+
+        @Override
+        public DataTypes.ArrayType defineArrayType(IDataType elemType) throws MetadataException {
+            return TypeSystem.this.defineArrayType(elemType);
+        }
+
+        @Override
+        public DataTypes.MapType defineMapType(IDataType keyType, IDataType valueType) throws MetadataException {
+            return TypeSystem.this.defineMapType(keyType, valueType);
+        }
     }
 }

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

@@ -20,6 +20,8 @@ package org.apache.metadata.types;
 import org.apache.metadata.MetadataException;
 
 import java.io.IOException;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 public class TypeUtils {
 
@@ -30,4 +32,28 @@ public class TypeUtils {
             throw new MetadataException(ie);
         }
     }
+
+    public static final String NAME_REGEX = "[a-zA-z][a-zA-Z0-9_]*";
+    public static final Pattern NAME_PATTERN = Pattern.compile(NAME_REGEX);
+
+    public static void validateName(String name) throws MetadataException {
+        if ( !NAME_PATTERN.matcher(name).matches() ) {
+            throw new MetadataException(String.format("Unsupported name for an attribute '%s'", name));
+        }
+    }
+
+    public static final Pattern ARRAY_TYPE_NAME_PATTERN = Pattern.compile(String.format("array<(%s)>", NAME_REGEX));
+    public static final Pattern MAP_TYPE_NAME_PATTERN =
+            Pattern.compile(String.format("map<(%s),(%s)>", NAME_REGEX, NAME_REGEX));
+
+    public static String parseAsArrayType(String typeName) {
+        Matcher m = ARRAY_TYPE_NAME_PATTERN.matcher(typeName);
+        return m.matches() ?  m.group(1) : null;
+    }
+
+    public static String[] parseAsMapType(String typeName) {
+        Matcher m = MAP_TYPE_NAME_PATTERN.matcher(typeName);
+        return m.matches() ?  new String[] {m.group(1), m.group(2)} : null;
+    }
+
 }

src/main/java/org/apache/metadata/types/TypedStructHandler.java

@@ -24,6 +24,8 @@ import org.apache.metadata.IStruct;
 import org.apache.metadata.ITypedStruct;
 import org.apache.metadata.MetadataException;
 import org.apache.metadata.Struct;
+import org.apache.metadata.storage.Id;
+import org.apache.metadata.storage.ReferenceableInstance;
 import org.apache.metadata.storage.StructInstance;
 
 import java.math.BigDecimal;
@@ -93,7 +95,9 @@ public class TypedStructHandler {
                 fieldMapping.numStrings == 0 ? null : new String[fieldMapping.numStrings],
                 fieldMapping.numArrays == 0 ? null : new ImmutableList[fieldMapping.numArrays],
                 fieldMapping.numMaps == 0 ? null : new ImmutableMap[fieldMapping.numMaps],
-                fieldMapping.numStructs == 0 ? null : new StructInstance[fieldMapping.numStructs]);
+                fieldMapping.numStructs == 0 ? null : new StructInstance[fieldMapping.numStructs],
+                fieldMapping.numReferenceables == 0 ? null : new ReferenceableInstance[fieldMapping.numReferenceables],
+                fieldMapping.numReferenceables == 0 ? null : new Id[fieldMapping.numReferenceables]);
     }
 
     public void output(IStruct s, Appendable buf, String prefix) throws MetadataException {

src/main/scala/org/apache/metadata/dsl/DynamicTypedStruct.scala

@@ -20,7 +20,7 @@ package org.apache.metadata.dsl
 
 import org.apache.metadata.{ITypedStruct, MetadataService}
 import org.apache.metadata.storage.StructInstance
-import org.apache.metadata.types.TypeSystem
+import org.apache.metadata.types.{StructType, TypeSystem}
 import scala.language.dynamics
 
 class DynamicTypedStruct(val ts : ITypedStruct) extends Dynamic {
@@ -32,5 +32,5 @@ class DynamicTypedStruct(val ts : ITypedStruct) extends Dynamic {
     }
     ts.set(name, value1)
   }
-  def dataType = MetadataService.getCurrentTypeSystem.getDataType(ts.getTypeName)
+  def dataType = MetadataService.getCurrentTypeSystem.getDataType(classOf[StructType], ts.getTypeName)
 }

src/main/scala/org/apache/metadata/dsl/package.scala

@@ -82,7 +82,7 @@ package object dsl {
 
   def createInstance(typeName : String) = {
     new DynamicTypedStruct(
-      ts.getDataType(typeName).asInstanceOf[IConstructableType[IStruct, ITypedStruct]].createInstance())
+      ts.getDataType(classOf[StructType],typeName).asInstanceOf[IConstructableType[IStruct, ITypedStruct]].createInstance())
   }
 
   implicit def dynTypedStructToTypedStruct(s : DynamicTypedStruct) = s.ts

src/main/scala/org/apache/metadata/json/Serialization.scala

@@ -76,7 +76,8 @@ class TypedStructSerializer extends Serializer[ITypedStruct] {
       case JObject(fs) =>
         val(typ, fields) = fs.partition(f => f._1 == Serialization.STRUCT_TYPE_FIELD_NAME)
         val typName = typ(0)._2.asInstanceOf[JString].s
-        val sT = MetadataService.getCurrentTypeSystem().getDataType(typName).asInstanceOf[IConstructableType[IStruct, ITypedStruct]]
+        val sT = MetadataService.getCurrentTypeSystem().getDataType(
+          classOf[IConstructableType[IStruct, ITypedStruct]], typName).asInstanceOf[IConstructableType[IStruct, ITypedStruct]]
         val s = sT.createInstance()
         fields.foreach { f =>
           val fName = f._1

src/test/java/org/apache/metadata/BaseTest.java

@@ -75,7 +75,7 @@ public abstract class BaseTest {
     }
 
     public static Struct createStruct(MetadataService ms) throws MetadataException {
-        StructType structType = (StructType) ms.getTypeSystem().getDataType(STRUCT_TYPE_1);
+        StructType structType = (StructType) ms.getTypeSystem().getDataType(StructType.class, STRUCT_TYPE_1);
         Struct s = new Struct(structType.getName());
         s.set("a", 1);
         s.set("b", true);
@@ -135,6 +135,9 @@ public abstract class BaseTest {
         return new TraitTypeDefinition(name, superTypes, attrDefs);
     }
 
-
+    protected ClassTypeDefinition createClassTypeDef(String name, ImmutableList<String> superTypes,
+                                                     AttributeDefinition... attrDefs) {
+        return new ClassTypeDefinition(name, superTypes, attrDefs);
+    }
 
 }

src/test/java/org/apache/metadata/ClassTest.java

+package org.apache.metadata;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.metadata.types.*;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ClassTest extends BaseTest {
+
+
+    @Before
+    public void setup() throws MetadataException {
+        super.setup();
+    }
+
+    /*
+     * Class Hierarchy is:
+     *   Department(name, employee : Array[Person])
+     *   Person(name, Department, Manager)
+     *   Manager(subordinate : Array[Person]) extends Person
+     *
+     * Persons can have SecurityClearance(level : Int) clearance.
+     */
+    @Test
+    public void test1() throws MetadataException {
+
+        TypeSystem ts = ms.getTypeSystem();
+
+        /*ClassTypeDefinition deptDef = createClassTypeDef("Department", ImmutableList.<String>of(),
+                createRequiredAttrDef("name", DataTypes.STRING_TYPE),
+                createOptionalAttrDef("employee", ts.defineArrayType(DataTypes.INT_TYPE)))*/
+    }
+}

src/test/java/org/apache/metadata/StructTest.java

@@ -32,8 +32,8 @@ public class StructTest extends BaseTest {
     @Before
     public void setup() throws MetadataException {
         super.setup();
-        structType = (StructType) ms.getTypeSystem().getDataType(STRUCT_TYPE_1);
-        recursiveStructType = (StructType) ms.getTypeSystem().getDataType(STRUCT_TYPE_2);
+        structType = (StructType) ms.getTypeSystem().getDataType(StructType.class, STRUCT_TYPE_1);
+        recursiveStructType = (StructType) ms.getTypeSystem().getDataType(StructType.class, STRUCT_TYPE_2);
     }
 
     @Test

src/test/java/org/apache/metadata/TraitTest.java

@@ -48,7 +48,7 @@ public class TraitTest extends BaseTest {
 
         defineTraits(A, B, C, D);
 
-        TraitType DType = (TraitType) ms.getTypeSystem().getDataType("D");
+        TraitType DType = (TraitType) ms.getTypeSystem().getDataType(TraitType.class, "D");
 
         Struct s1 = new Struct("D");
         s1.set("d", 1);
@@ -84,7 +84,7 @@ public class TraitTest extends BaseTest {
         /*
          * cast to B and set the 'b' attribute on A.
          */
-        TraitType BType = (TraitType) ms.getTypeSystem().getDataType("B");
+        TraitType BType = (TraitType) ms.getTypeSystem().getDataType(TraitType.class, "B");
         IStruct s2 = DType.castAs(ts, "B");
         s2.set("A.B.b", false);
 
@@ -106,7 +106,7 @@ public class TraitTest extends BaseTest {
         /*
          * cast again to A and set the 'b' attribute on A.
          */
-        TraitType AType = (TraitType) ms.getTypeSystem().getDataType("A");
+        TraitType AType = (TraitType) ms.getTypeSystem().getDataType(TraitType.class, "A");
         IStruct s3 = BType.castAs(s2, "A");
         s3.set("b", true);
         Assert.assertEquals(ts.toString(), "{\n" +

src/test/scala/org/apache/metadata/json/SerializationTest.scala

@@ -40,8 +40,8 @@ class SerializationTest extends BaseTest {
   @Before
   override def setup {
     super.setup
-    structType = ms.getTypeSystem.getDataType(BaseTest.STRUCT_TYPE_1).asInstanceOf[StructType]
-    recursiveStructType = ms.getTypeSystem.getDataType(BaseTest.STRUCT_TYPE_2).asInstanceOf[StructType]
+    structType = ms.getTypeSystem.getDataType(classOf[StructType], BaseTest.STRUCT_TYPE_1).asInstanceOf[StructType]
+    recursiveStructType = ms.getTypeSystem.getDataType(classOf[StructType], BaseTest.STRUCT_TYPE_2).asInstanceOf[StructType]
   }
 
   @Test def test1 {
@@ -92,7 +92,7 @@ class SerializationTest extends BaseTest {
 
     defineTraits(A, B, C, D)
 
-    val DType: TraitType = ms.getTypeSystem.getDataType("D").asInstanceOf[TraitType]
+    val DType: TraitType = ms.getTypeSystem.getDataType(classOf[TraitType], "D").asInstanceOf[TraitType]
     val s1: Struct = new Struct("D")
     s1.set("d", 1)
     s1.set("c", 1)