add support for Traits

src/main/java/org/apache/metadata/storage/DownCastStructInstance.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 org.apache.metadata.IStruct;
+import org.apache.metadata.MetadataException;
+import org.apache.metadata.types.DownCastFieldMapping;
+
+public class DownCastStructInstance implements IStruct {
+
+    public final String typeName;
+    public final DownCastFieldMapping fieldMapping;
+    public final IStruct backingInstance;
+
+    public DownCastStructInstance(String typeName, DownCastFieldMapping fieldMapping,
+                                  IStruct backingInstance) {
+        this.typeName = typeName;
+        this.fieldMapping = fieldMapping;
+        this.backingInstance = backingInstance;
+    }
+
+    @Override
+    public String getTypeName() {
+        return typeName;
+    }
+
+    @Override
+    public Object get(String attrName) throws MetadataException {
+        return fieldMapping.get(this, attrName);
+    }
+
+    @Override
+    public void set(String attrName, Object val) throws MetadataException {
+        fieldMapping.set(this, attrName, val);
+    }
+}
+
+

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

@@ -28,6 +28,7 @@ import org.apache.metadata.types.TypeUtils;
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.util.Date;
+import java.util.Map;
 
 public class StructInstance implements IStruct {
     public final String dataTypeName;
@@ -100,9 +101,11 @@ public class StructInstance implements IStruct {
         }
         TypeUtils.outputVal("\n", buf, "");
         String fieldPrefix = prefix + "\t";
-        for(AttributeInfo i : fieldMapping.fields.values()) {
-            Object aVal = s.get(i.name);
-            TypeUtils.outputVal(i.name + " : ", buf, fieldPrefix);
+        for(Map.Entry<String,AttributeInfo> e : fieldMapping.fields.entrySet()) {
+            String attrName = e.getKey();
+            AttributeInfo i = e.getValue();
+            Object aVal = s.get(attrName);
+            TypeUtils.outputVal(attrName + " : ", buf, fieldPrefix);
             i.dataType().output(aVal, buf, "");
             TypeUtils.outputVal("\n", buf, "");
         }
@@ -118,9 +121,11 @@ public class StructInstance implements IStruct {
             TypeUtils.outputVal("{", buf, prefix);
             TypeUtils.outputVal("\n", buf, "");
             String fieldPrefix = prefix + "\t";
-            for(AttributeInfo i : fieldMapping.fields.values()) {
-                Object aVal = get(i.name);
-                TypeUtils.outputVal(i.name + " : ", buf, fieldPrefix);
+            for(Map.Entry<String,AttributeInfo> e : fieldMapping.fields.entrySet()) {
+                String attrName = e.getKey();
+                AttributeInfo i = e.getValue();
+                Object aVal = get(attrName);
+                TypeUtils.outputVal(attrName + " : ", buf, fieldPrefix);
                 i.dataType().output(aVal, buf, "");
                 TypeUtils.outputVal("\n", buf, "");
             }

src/main/java/org/apache/metadata/types/DownCastFieldMapping.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.ImmutableMap;
+import org.apache.metadata.MetadataException;
+import org.apache.metadata.storage.DownCastStructInstance;
+import org.apache.metadata.storage.StructInstance;
+
+public class DownCastFieldMapping {
+
+    public final ImmutableMap<String, String> fieldNameMap;
+
+    protected DownCastFieldMapping(ImmutableMap<String, String> fieldNameMap) {
+        this.fieldNameMap = fieldNameMap;
+    }
+
+    public void set(DownCastStructInstance s, String attrName, Object val) throws MetadataException {
+
+        String mappedNm = fieldNameMap.get(attrName);
+        if ( mappedNm == null ) {
+            throw new ValueConversionException(s.getTypeName(), val, "Unknown field " + attrName);
+        }
+
+        s.backingInstance.set(mappedNm, val);
+    }
+
+    public Object get(DownCastStructInstance s, String attrName) throws MetadataException {
+
+        String mappedNm = fieldNameMap.get(attrName);
+        if ( mappedNm == null ) {
+            throw new ValueConversionException(
+                    String.format("Unknown field %s for Struct %s", attrName, s.getTypeName()));
+        }
+        return s.backingInstance.get(mappedNm);
+    }
+}

src/main/java/org/apache/metadata/types/ITypeBrowser.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;
+
+/**
+ * @nopublic
+ */
+public interface ITypeBrowser {
+
+    IDataType dataType(String name);
+}

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

@@ -30,20 +30,28 @@ import java.util.*;
 
 public class StructType  extends AbstractDataType<IStruct> {
 
+    public final ITypeBrowser typeSystem;
     public final String name;
     public final FieldMapping fieldMapping;
+    public final int numFields;
 
     /**
      * Used when creating a StructType, to support recursive Structs.
      */
-    StructType(String name) {
+    protected StructType(ITypeBrowser typeSystem, String name, int numFields) {
+        this.typeSystem = typeSystem;
         this.name = name;
         this.fieldMapping = null;
+        this.numFields = numFields;
     }
 
-    StructType(String name, AttributeInfo... fields) throws MetadataException {
+    protected StructType(ITypeBrowser typeSystem, String name,
+                         ImmutableList<String> superTypes, AttributeInfo... fields) throws MetadataException {
+        this.typeSystem = typeSystem;
         this.name = name;
-        this.fieldMapping = constructFieldMapping(fields);
+        this.fieldMapping = constructFieldMapping(superTypes,
+                fields);
+        this.numFields = this.fieldMapping.fields.size();
     }
 
     @Override
@@ -51,7 +59,8 @@ public class StructType  extends AbstractDataType<IStruct> {
         return name;
     }
 
-    protected FieldMapping constructFieldMapping(AttributeInfo... fields)
+    protected FieldMapping constructFieldMapping(ImmutableList<String> superTypes,
+                                                 AttributeInfo... fields)
             throws MetadataException {
 
         Map<String,AttributeInfo> fieldsMap = new LinkedHashMap<String, AttributeInfo>();
@@ -155,10 +164,12 @@ public class StructType  extends AbstractDataType<IStruct> {
                     throw new ValueConversionException(this, val);
                 }
                 StructInstance ts = createInstance();
-                for(AttributeInfo i : fieldMapping.fields.values()) {
-                    Object aVal = s.get(i.name);
+                for(Map.Entry<String,AttributeInfo> e : fieldMapping.fields.entrySet() ) {
+                    String attrKey = e.getKey();
+                    AttributeInfo i = e.getValue();
+                    Object aVal = s.get(attrKey);
                     try {
-                        ts.set(i.name, aVal);
+                        ts.set(attrKey, aVal);
                     } catch(ValueConversionException ve) {
                         throw new ValueConversionException(this, val, ve);
                     }

src/main/java/org/apache/metadata/types/TraitType.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.IStruct;
+import org.apache.metadata.MetadataException;
+import org.apache.metadata.Struct;
+import org.apache.metadata.storage.DownCastStructInstance;
+import org.apache.metadata.storage.StructInstance;
+
+import java.util.*;
+
+public class TraitType extends StructType implements Comparable<TraitType> {
+
+    public final ImmutableList<String> superTraits;
+    protected final ImmutableList<AttributeInfo> immediateAttrs;
+    protected ImmutableMap<String, List<Path>> superTypePaths;
+    protected ImmutableMap<String, Path> pathNameToPathMap;
+
+    /**
+     * Used when creating a TraitType, to support recursive Structs.
+     */
+    TraitType(ITypeBrowser typeSystem, String name, ImmutableList<String> superTraits, int numFields) {
+        super(typeSystem, name, numFields);
+        this.superTraits = superTraits;
+        this.immediateAttrs = null;
+    }
+
+    TraitType(ITypeBrowser typeSystem, String name, ImmutableList<String> superTraits, AttributeInfo... fields)
+            throws MetadataException {
+        super(typeSystem, name, superTraits, fields);
+        this.superTraits = superTraits == null ? ImmutableList.<String>of() : superTraits;
+        this.immediateAttrs = ImmutableList.<AttributeInfo>copyOf(fields);
+    }
+
+    @Override
+    public String getName() {
+        return name;
+    }
+
+    @Override
+    protected FieldMapping constructFieldMapping(ImmutableList<String> superTraits,
+                                                 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>();
+        int numBools = 0;
+        int numBytes = 0;
+        int numShorts = 0;
+        int numInts = 0;
+        int numLongs = 0;
+        int numFloats = 0;
+        int numDoubles = 0;
+        int numBigInts = 0;
+        int numBigDecimals = 0;
+        int numDates = 0;
+        int numStrings = 0;
+        int numArrays = 0;
+        int numMaps = 0;
+        int numStructs = 0;
+
+        Map<String, List<Path>> superTypePaths = new HashMap<String, List<Path>>();
+        Map<String, Path> pathNameToPathMap = new HashMap<String, Path>();
+        Queue<Path> queue = new LinkedList<Path>();
+        queue.add(new Node(getName()));
+        while(!queue.isEmpty()) {
+            Path currentPath = queue.poll();
+            TraitType superType = currentPath.typeName == getName() ? this :
+                    (TraitType) typeSystem.dataType(currentPath.typeName);
+
+            pathNameToPathMap.put(currentPath.pathName, currentPath);
+            if ( superType != this ) {
+                List<Path> typePaths = superTypePaths.get(superType.getName());
+                if ( typePaths == null ) {
+                    typePaths = new ArrayList<Path>();
+                    superTypePaths.put(superType.getName(), typePaths);
+                }
+                typePaths.add(currentPath);
+            }
+
+            ImmutableList<AttributeInfo> superTypeFields = superType == this ?
+                    ImmutableList.<AttributeInfo>copyOf(fields) : superType.immediateAttrs;
+
+            Set<String> immediateFields = new HashSet<String>();
+
+            for(AttributeInfo i : superTypeFields) {
+                if ( superType == this && immediateFields.contains(i.name) ) {
+                    throw new MetadataException(
+                            String.format("Struct defintion cannot contain multiple fields with the same name %s",
+                                    i.name));
+                }
+                
+                String attrName = i.name;
+                if ( fieldsMap.containsKey(attrName)) {
+                    attrName = currentPath.addOverrideAttr(attrName);
+                }
+                
+                fieldsMap.put(attrName, i);
+                fieldNullPos.put(attrName, fieldNullPos.size());
+                if ( i.dataType() == DataTypes.BOOLEAN_TYPE ) {
+                    fieldPos.put(attrName, numBools);
+                    numBools++;
+                } else if ( i.dataType() == DataTypes.BYTE_TYPE ) {
+                    fieldPos.put(attrName, numBytes);
+                    numBytes++;
+                } else if ( i.dataType() == DataTypes.SHORT_TYPE ) {
+                    fieldPos.put(attrName, numShorts);
+                    numShorts++;
+                } else if ( i.dataType() == DataTypes.INT_TYPE ) {
+                    fieldPos.put(attrName, numInts);
+                    numInts++;
+                } else if ( i.dataType() == DataTypes.LONG_TYPE ) {
+                    fieldPos.put(attrName, numLongs);
+                    numLongs++;
+                } else if ( i.dataType() == DataTypes.FLOAT_TYPE ) {
+                    fieldPos.put(attrName, numFloats);
+                    numFloats++;
+                } else if ( i.dataType() == DataTypes.DOUBLE_TYPE ) {
+                    fieldPos.put(attrName, numDoubles);
+                    numDoubles++;
+                } else if ( i.dataType() == DataTypes.BIGINTEGER_TYPE ) {
+                    fieldPos.put(attrName, numBigInts);
+                    numBigInts++;
+                } else if ( i.dataType() == DataTypes.BIGDECIMAL_TYPE ) {
+                    fieldPos.put(attrName, numBigDecimals);
+                    numBigDecimals++;
+                } else if ( i.dataType() == DataTypes.DATE_TYPE ) {
+                    fieldPos.put(attrName, numDates);
+                    numDates++;
+                } else if ( i.dataType() == DataTypes.STRING_TYPE ) {
+                    fieldPos.put(attrName, numStrings);
+                    numStrings++;
+                } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.ARRAY ) {
+                    fieldPos.put(attrName, numArrays);
+                    numArrays++;
+                } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.MAP ) {
+                    fieldPos.put(attrName, numMaps);
+                    numMaps++;
+                } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.STRUCT ) {
+                    fieldPos.put(attrName, numStructs);
+                    numStructs++;
+                } else {
+                    throw new MetadataException(String.format("Unknown datatype %s", i.dataType()));
+                }
+            }
+
+            for(String sT : superType == this ? superTraits : superType.superTraits) {
+                queue.add(new Path(sT, currentPath));
+            }
+        }
+
+        this.superTypePaths = ImmutableMap.copyOf(superTypePaths);
+        this.pathNameToPathMap = ImmutableMap.copyOf(pathNameToPathMap);
+
+        return new FieldMapping(fieldsMap,
+                fieldPos,
+                fieldNullPos,
+                numBools,
+                numBytes,
+                numShorts,
+                numInts,
+                numLongs,
+                numFloats,
+                numDoubles,
+                numBigInts,
+                numBigDecimals,
+                numDates,
+                numStrings,
+                numArrays,
+                numMaps,
+                numStructs);
+    }
+
+    @Override
+    public DataTypes.TypeCategory getTypeCategory() {
+        return DataTypes.TypeCategory.TRAIT;
+    }
+
+    protected Map<String, String> constructDowncastFieldMap(TraitType subType, Path pathToSubType) {
+
+        String pathToSubTypeName = pathToSubType.pathAfterThis;
+        /*
+         * the downcastMap;
+         */
+        Map<String, String> dCMap = new HashMap<String, String>();
+        Iterator<Path> itr = pathIterator();
+        while(itr.hasNext()) {
+            Path p = itr.next();
+            Path pInSubType = subType.pathNameToPathMap.get(p.pathName + "." + pathToSubTypeName);
+
+            if ( pInSubType.hiddenAttributeMap != null ) {
+                for(Map.Entry<String, String> e : pInSubType.hiddenAttributeMap.entrySet()) {
+                    String mappedInThisType =
+                            p.hiddenAttributeMap != null ? p.hiddenAttributeMap.get(e.getKey()) : null;
+                    if ( mappedInThisType == null ) {
+                        dCMap.put(e.getKey(), e.getValue());
+                    } else {
+                        dCMap.put(mappedInThisType, e.getValue());
+                    }
+                }
+            }
+        }
+        return dCMap;
+    }
+
+    public IStruct castAs(IStruct s, String superTypeName) throws MetadataException {
+
+        if ( !superTypePaths.containsKey(superTypeName) ) {
+            throw new MetadataException(String.format("Cannot downcast to %s from type %s", superTypeName, getName()));
+        }
+
+        if (s != null) {
+            if (s.getTypeName() != getName()) {
+                throw new MetadataException(
+                        String.format("Downcast called on wrong type %s, instance type is %s",
+                                getName(), s.getTypeName()));
+            }
+
+            List<Path> pathToSuper = superTypePaths.get(superTypeName);
+            if ( pathToSuper.size() > 1 ) {
+                throw new MetadataException(
+                        String.format("Cannot downcast called to %s, from %s: there are multiple paths to SuperType",
+                                superTypeName, getName()));
+            }
+
+            TraitType superType = (TraitType) typeSystem.dataType(superTypeName);
+            Map<String, String> downCastMap = superType.constructDowncastFieldMap(this, pathToSuper.get(0));
+            return new DownCastStructInstance(superTypeName,
+                    new DownCastFieldMapping(ImmutableMap.copyOf(downCastMap)),
+                    s);
+        }
+
+        return null;
+    }
+
+    public Iterator<Path> pathIterator() {
+        return new PathItr();
+    }
+
+    @Override
+    public int compareTo(TraitType o) {
+        String oName = o.getName();
+        if ( superTraits.contains(oName) ) {
+            return 1;
+        } else if ( o.superTraits.contains(getName())) {
+            return -1;
+        } else {
+            return getName().compareTo(oName);
+        }
+    }
+
+    class PathItr implements Iterator<Path> {
+
+        Queue<Path> pathQueue;
+
+        PathItr() {
+            pathQueue = new LinkedList<Path>();
+            pathQueue.add(pathNameToPathMap.get(getName()));
+        }
+
+        @Override
+        public boolean hasNext() {
+            return !pathQueue.isEmpty();
+        }
+
+        @Override
+        public Path next() {
+            Path p = pathQueue.poll();
+            TraitType t = (TraitType) typeSystem.dataType(p.typeName);
+            if ( t.superTraits != null ) {
+                for(String sT : t.superTraits) {
+                    String nm = sT + "." + p.pathName;
+                    pathQueue.add(pathNameToPathMap.get(nm));
+                }
+            }
+            return p;
+        }
+
+        @Override
+        public void remove() {
+            throw new UnsupportedOperationException();
+        }
+    }
+
+    static class Path {
+        public final String typeName;
+        private final Path subTypePath;
+        public final String pathName;
+        public final String pathAfterThis;
+        /*
+         * name mapping for attributes hidden by a SubType.
+         */
+        Map<String, String> hiddenAttributeMap;
+
+        Path(String typeName, Path childPath) throws MetadataException {
+            this.typeName = typeName;
+            this.subTypePath = childPath;
+            if ( childPath.contains(typeName) ) {
+                throw new CyclicTypeDefinition(this);
+            }
+            pathName = String.format("%s.%s", typeName, childPath.pathName);
+            pathAfterThis = childPath.pathName;
+        }
+
+        Path(String typeName) {
+            assert getClass() == Node.class;
+            this.typeName = typeName;
+            this.subTypePath = null;
+            pathName = typeName;
+            pathAfterThis = null;
+        }
+
+        public boolean contains(String typeName) {
+            return this.typeName.equals(typeName) || (subTypePath != null && subTypePath.contains(typeName));
+        }
+
+        public String pathString(String nodeSep) {
+
+            StringBuilder b = new StringBuilder();
+            Path p = this;
+
+            while ( p != null ) {
+                b.append(p.typeName);
+                p = p.subTypePath;
+                if ( p != null ) {
+                    b.append(nodeSep);
+                }
+            }
+            return b.toString();
+        }
+
+        String addOverrideAttr(String name) {
+            hiddenAttributeMap = hiddenAttributeMap == null ? new HashMap<String, String>() : hiddenAttributeMap;
+            String oName = pathName + "." + name;
+            hiddenAttributeMap.put(name, oName);
+            return oName;
+        }
+    }
+
+    static class Node extends Path {
+        Node(String typeName) {
+            super(typeName);
+        }
+    }
+
+    static class CyclicTypeDefinition extends MetadataException {
+
+        CyclicTypeDefinition(Path p) {
+            super(String.format("Cycle in Type Definition %s", p.pathString(" -> ")));
+        }
+    }
+}

src/main/java/org/apache/metadata/types/TraitTypeDefinition.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.ImmutableList;
+
+public class TraitTypeDefinition {
+
+    public final String typeName;
+    public final ImmutableList<String> superTraits;
+    public final AttributeDefinition[] attributeDefinitions;
+
+    public TraitTypeDefinition(String typeName, ImmutableList<String> superTraits,
+                               AttributeDefinition[] attributeDefinitions) {
+        this.typeName = typeName;
+        this.superTraits = superTraits == null ? ImmutableList.<String>of() : superTraits;
+        this.attributeDefinitions = attributeDefinitions;
+    }
+}

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

@@ -15,15 +15,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.apache.metadata.types;
 
 import com.google.common.collect.ImmutableList;
 import org.apache.metadata.MetadataException;
 
-import java.util.HashMap;
-import java.util.Map;
+import java.util.*;
 
-public class TypeSystem {
+public class TypeSystem implements ITypeBrowser {
 
     private Map<String, IDataType> types;
 
@@ -50,6 +50,10 @@ public class TypeSystem {
         types.put(DataTypes.STRING_TYPE.getName(), DataTypes.STRING_TYPE);
     }
 
+    public IDataType dataType(String name) {
+        return types.get(name);
+    }
+
     public IDataType getDataType(String name) throws MetadataException {
         if ( types.containsKey(name) ) {
             return types.get(name);
@@ -67,7 +71,7 @@ public class TypeSystem {
         AttributeInfo[] infos = new AttributeInfo[attrDefs.length];
         Map<Integer, AttributeDefinition> recursiveRefs = new HashMap<Integer, AttributeDefinition>();
         try {
-            types.put(name, new StructType(name));
+            types.put(name, new StructType(this, name, attrDefs.length));
             for (int i = 0; i < attrDefs.length; i++) {
                 infos[i] = new AttributeInfo(this, attrDefs[i]);
                 if ( attrDefs[i].dataTypeName == name ) {
@@ -81,7 +85,7 @@ public class TypeSystem {
             types.remove(name);
             throw re;
         }
-        StructType sT = new StructType(name, infos);
+        StructType sT = new StructType(this, name, null, infos);
         types.put(name, sT);
         for(Map.Entry<Integer, AttributeDefinition> e : recursiveRefs.entrySet()) {
             infos[e.getKey()].setDataType(sT);
@@ -89,6 +93,109 @@ public class TypeSystem {
         return sT;
     }
 
+    public TraitType defineTraitType(boolean errorIfExists,
+                                     TraitTypeDefinition traitDef
+                                       ) throws MetadataException {
+        Map<String, TraitType> m = defineTraitTypes(errorIfExists, traitDef);
+        return m.values().iterator().next();
+    }
+
+    public Map<String, TraitType> defineTraitTypes(boolean errorIfExists,
+                                                   TraitTypeDefinition... traitDefs
+                                      ) throws MetadataException {
+        TransientTypeSystem transientTypes = new TransientTypeSystem();
+        Map<String,TraitTypeDefinition> traitDefMap = new HashMap<String, TraitTypeDefinition>();
+
+
+        /*
+         * Step 1:
+         * - validate cannot redefine types
+         * - setup an empty TraitType to allow for recursive type graphs.
+         */
+        for(TraitTypeDefinition traitDef : traitDefs) {
+            assert traitDef.typeName != null;
+            if ( types.containsKey(traitDef.typeName) ) {
+                throw new MetadataException(String.format("Cannot redefine type %s", traitDef.typeName));
+            }
+
+            transientTypes.traitTypes.put(traitDef.typeName,
+                    new TraitType(transientTypes, traitDef.typeName, traitDef.superTraits,
+                    traitDef.attributeDefinitions.length));
+            traitDefMap.put(traitDef.typeName, traitDef);
+        }
+
+        /*
+         * Step 2:
+         * - validate SuperTypes.
+         */
+        for(TraitTypeDefinition traitDef : traitDefs) {
+            Set<String> s = new HashSet<String>();
+            for(String superTraitName : traitDef.superTraits ) {
+
+                if (s.contains(superTraitName) ) {
+                    throw new MetadataException(String.format("Trait %s extends superTrait %s multiple times",
+                            traitDef.typeName, superTraitName));
+                }
+
+                IDataType dT = types.get(superTraitName);
+                dT = dT == null ? transientTypes.traitTypes.get(superTraitName) : dT;
+
+                if ( dT == null ) {
+                    throw new MetadataException(String.format("Unknown superType %s in definition of type %s",
+                            superTraitName, traitDef.typeName));
+                }
+
+                if ( dT.getTypeCategory() != DataTypes.TypeCategory.TRAIT ) {
+                    throw new MetadataException(String.format("SuperType %s must be a Trait, in definition of type %s",
+                            superTraitName, traitDef.typeName));
+                }
+                s.add(superTraitName);
+            }
+        }
+
+        /*
+         * Step 3:
+         * - Construct TraitTypes in order of SuperType before SubType.
+         */
+        List<TraitType> l = new ArrayList<TraitType>(transientTypes.traitTypes.values());
+        Collections.sort(l);
+        List<AttributeInfo> recursiveRefs = new ArrayList<AttributeInfo>();
+
+        try {
+            for (TraitType ttO : l) {
+                TraitTypeDefinition traitDef = traitDefMap.get(ttO.getName());
+                AttributeInfo[] infos = new AttributeInfo[traitDef.attributeDefinitions.length];
+                for (int i = 0; i < traitDef.attributeDefinitions.length; i++) {
+                    infos[i] = new AttributeInfo(this, traitDef.attributeDefinitions[i]);
+                    if (transientTypes.traitTypes.containsKey(traitDef.attributeDefinitions[i].dataTypeName)) {
+                        recursiveRefs.add(infos[i]);
+                    }
+                }
+
+                TraitType tt = new TraitType(this, traitDef.typeName, traitDef.superTraits, infos);
+                ;
+                types.put(tt.getName(), tt);
+            }
+
+            /*
+             * Step 4:
+             * - fix up references in recursive AttrInfo
+             */
+            for (AttributeInfo info : recursiveRefs) {
+                info.setDataType(dataType(info.dataType().getName()));
+            }
+        } catch(MetadataException me) {
+            for(String sT : transientTypes.traitTypes.keySet()) {
+                types.remove(sT);
+            }
+            throw me;
+        }
+
+        return transientTypes.traitTypes;
+    }
+
+
+
     public DataTypes.ArrayType defineArrayType(IDataType elemType) throws MetadataException {
         assert elemType != null;
         DataTypes.ArrayType dT = new DataTypes.ArrayType(elemType);
@@ -103,4 +210,14 @@ public class TypeSystem {
         types.put(dT.getName(), dT);
         return dT;
     }
+
+    class TransientTypeSystem implements ITypeBrowser {
+        Map<String, TraitType> traitTypes = new HashMap<String, TraitType>();
+
+        public IDataType dataType(String name) {
+            IDataType dT = TypeSystem.this.dataType(name);
+            dT = dT == null ? traitTypes.get(name) : dT;
+            return dT;
+        }
+    }
 }

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

@@ -18,6 +18,7 @@
 
 package org.apache.metadata;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import junit.framework.TestCase;
@@ -40,11 +41,13 @@ public abstract class BaseTest {
     public static final String STRUCT_TYPE_1 = "t1";
     public static final String STRUCT_TYPE_2 = "t2";
 
-    @BeforeClass
-    public static void setupClass() throws MetadataException {
+    @Before
+    public void setup() throws MetadataException {
+
         TypeSystem ts = new TypeSystem();
         MemRepository mr = new MemRepository();
-        MetadataService.setCurrentService(new MetadataService(mr, ts));
+        ms = new MetadataService(mr, ts);
+        MetadataService.setCurrentService(ms);
 
         StructType structType = ts.defineStructType(STRUCT_TYPE_1,
                 true,
@@ -95,11 +98,6 @@ public abstract class BaseTest {
         return s;
     }
 
-    @Before
-    public void setup() throws MetadataException {
-        ms = MetadataService.getCurrentService();
-    }
-
     public static AttributeDefinition createOptionalAttrDef(String name,
                                                             IDataType dataType
     ) {
@@ -128,4 +126,15 @@ public abstract class BaseTest {
         return new AttributeDefinition(name, dataType, Multiplicity.REQUIRED, false, null);
     }
 
+    protected Map<String, TraitType> defineTraits(TraitTypeDefinition... tDefs) throws MetadataException {
+        return ms.getTypeSystem().defineTraitTypes(true, tDefs);
+    }
+
+    protected TraitTypeDefinition createTraitTypeDef(String name, ImmutableList<String> superTypes,
+                                        AttributeDefinition... attrDefs) {
+        return new TraitTypeDefinition(name, superTypes, attrDefs);
+    }
+
+
+
 }

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

+package org.apache.metadata;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.metadata.storage.StructInstance;
+import org.apache.metadata.types.*;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Map;
+
+public class TraitTest extends BaseTest {
+
+
+    @Before
+    public void setup() throws MetadataException {
+        super.setup();
+    }
+
+    /*
+     * Type Hierarchy is:
+     *   A(a,b,c,d)
+     *   B(b) extends A
+     *   C(c) extends A
+     *   D(d) extends B,C
+     *
+     * - There are a total of 11 fields in an instance of D
+     * - an attribute that is hidden by a SubType can referenced by prefixing it with the complete Path.
+     *   For e.g. the 'b' attribute in A (that is a superType for B) is hidden the 'b' attribute in B.
+     *   So it is availabel by the name 'A.B.D.b'
+     *
+     * - Another way to set attributes is to cast. Casting a 'D' instance of 'B' makes the 'A.B.D.b' attribute
+     *   available as 'A.B.b'. Casting one more time to an 'A' makes the 'A.B.b' attribute available as 'b'.
+     */
+    @Test
+    public void test1() throws MetadataException {
+        TraitTypeDefinition A = createTraitTypeDef("A", null,
+                createRequiredAttrDef("a", DataTypes.INT_TYPE),
+                createOptionalAttrDef("b", DataTypes.BOOLEAN_TYPE),
+                createOptionalAttrDef("c", DataTypes.BYTE_TYPE),
+                createOptionalAttrDef("d", DataTypes.SHORT_TYPE));
+        TraitTypeDefinition B = createTraitTypeDef("B", ImmutableList.<String>of("A"),
+                createOptionalAttrDef("b", DataTypes.BOOLEAN_TYPE));
+        TraitTypeDefinition C = createTraitTypeDef("C", ImmutableList.<String>of("A"),
+                createOptionalAttrDef("c", DataTypes.BYTE_TYPE));
+        TraitTypeDefinition D = createTraitTypeDef("D", ImmutableList.<String>of("B", "C"),
+                createOptionalAttrDef("d", DataTypes.SHORT_TYPE));
+
+        defineTraits(A, B, C, D);
+
+        TraitType DType = (TraitType) ms.getTypeSystem().getDataType("D");
+
+        Struct s1 = new Struct("D");
+        s1.set("d", 1);
+        s1.set("c", 1);
+        s1.set("b", true);
+        s1.set("a", 1);
+        s1.set("A.B.D.b", true);
+        s1.set("A.B.D.c", 2);
+        s1.set("A.B.D.d", 2);
+
+        s1.set("A.C.D.a", 3);
+        s1.set("A.C.D.b", false);
+        s1.set("A.C.D.c", 3);
+        s1.set("A.C.D.d", 3);
+
+
+        StructInstance ts = DType.convert(s1, Multiplicity.REQUIRED);
+        System.out.println(ts);
+
+        /*
+         * cast to B and set the 'b' attribute on A.
+         */
+        TraitType BType = (TraitType) ms.getTypeSystem().getDataType("B");
+        IStruct s2 = DType.castAs(ts, "B");
+        s2.set("A.B.b", false);
+
+        System.out.println(ts);
+
+        /*
+         * cast again to A and set the 'b' attribute on A.
+         */
+        TraitType AType = (TraitType) ms.getTypeSystem().getDataType("A");
+        IStruct s3 = BType.castAs(s2, "A");
+        s3.set("b", true);
+        System.out.println(ts);
+    }
+}
+

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

@@ -18,13 +18,12 @@
 
 package org.apache.metadata.json
 
+import com.google.common.collect.ImmutableList
 import org.apache.metadata.Struct
 import org.apache.metadata.storage.StructInstance
 import org.apache.metadata.storage.StructInstance
-import org.apache.metadata.types.Multiplicity
-import org.apache.metadata.types.StructType
+import org.apache.metadata.types._
 import org.apache.metadata.{Struct, BaseTest}
-import org.apache.metadata.types.{Multiplicity, StructType}
 import org.json4s.NoTypeHints
 import org.junit.Before
 import org.junit.Test
@@ -78,4 +77,52 @@ class SerializationTest extends BaseTest {
     println("Typed Struct read from string:")
     println(ts1)
   }
+
+  @Test def testTrait {
+    val A: TraitTypeDefinition = createTraitTypeDef("A", null,
+      BaseTest.createRequiredAttrDef("a", DataTypes.INT_TYPE),
+      BaseTest.createOptionalAttrDef("b", DataTypes.BOOLEAN_TYPE),
+      BaseTest.createOptionalAttrDef("c", DataTypes.BYTE_TYPE),
+      BaseTest.createOptionalAttrDef("d", DataTypes.SHORT_TYPE))
+    val B: TraitTypeDefinition = createTraitTypeDef("B", ImmutableList.of[String]("A"),
+      BaseTest.createOptionalAttrDef("b", DataTypes.BOOLEAN_TYPE))
+    val C: TraitTypeDefinition = createTraitTypeDef("C", ImmutableList.of[String]("A"),
+      BaseTest.createOptionalAttrDef("c", DataTypes.BYTE_TYPE))
+    val D: TraitTypeDefinition = createTraitTypeDef("D", ImmutableList.of[String]("B", "C"),
+      BaseTest.createOptionalAttrDef("d", DataTypes.SHORT_TYPE))
+
+    defineTraits(A, B, C, D)
+
+    val DType: TraitType = ms.getTypeSystem.getDataType("D").asInstanceOf[TraitType]
+    val s1: Struct = new Struct("D")
+    s1.set("d", 1)
+    s1.set("c", 1)
+    s1.set("b", true)
+    s1.set("a", 1)
+    s1.set("A.B.D.b", true)
+    s1.set("A.B.D.c", 2)
+    s1.set("A.B.D.d", 2)
+    s1.set("A.C.D.a", 3)
+    s1.set("A.C.D.b", false)
+    s1.set("A.C.D.c", 3)
+    s1.set("A.C.D.d", 3)
+
+    val s: Struct = BaseTest.createStruct(ms)
+    val ts: StructInstance = DType.convert(s1, Multiplicity.REQUIRED)
+
+    implicit val formats = org.json4s.native.Serialization.formats(NoTypeHints) + new TypedStructSerializer +
+      new BigDecimalSerializer + new BigIntegerSerializer
+
+    println("Typed Struct :")
+    println(ts)
+    val ser = swrite(ts)
+    println("Json representation :")
+    println(ser)
+    val ts1 = read[StructInstance](
+      """
+        {"$typeName$":"D","A.C.D.d":3,"A.B.D.c":2,"b":true,"A.C.D.c":3,"d":1,
+        "A.B.D.b":true,"a":1,"A.C.D.b":false,"A.B.D.d":2,"c":1,"A.C.D.a":3}""")
+    println("Typed Struct read from string:")
+    println(ts1)
+  }
 }