add support for EnumTypes

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

@@ -106,7 +106,7 @@ public class StructInstance implements ITypedStruct {
         if (val != null && val instanceof Id) {
             ClassType clsType =
                     MetadataService.getCurrentTypeSystem().getDataType(ClassType.class, i.dataType().getName());
-            clsType.validateId((Id)cVal);
+            clsType.validateId((Id)val);
             cVal = val;
         } else {
             cVal = i.dataType().convert(val, i.multiplicity);
@@ -138,6 +138,8 @@ public class StructInstance implements ITypedStruct {
             dates[pos] = (Date) cVal;
         } else if ( i.dataType() == DataTypes.STRING_TYPE ) {
             strings[pos] = (String) cVal;
+        } else if (i.dataType().getTypeCategory() == DataTypes.TypeCategory.ENUM ) {
+            ints[pos] = ((EnumValue)cVal).ordinal;
         } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.ARRAY ) {
             arrays[pos] = (ImmutableList) cVal;
         } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.MAP ) {
@@ -190,6 +192,8 @@ public class StructInstance implements ITypedStruct {
             return dates[pos];
         } else if ( i.dataType() == DataTypes.STRING_TYPE ) {
             return strings[pos];
+        } else if (i.dataType().getTypeCategory() == DataTypes.TypeCategory.ENUM ) {
+            return ((EnumType)i.dataType()).fromOrdinal(ints[pos]);
         } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.ARRAY ) {
             return arrays[pos];
         } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.MAP ) {

typesystem/src/main/java/org/apache/hadoop/metadata/storage/memory/AttributeStores.java

@@ -79,6 +79,8 @@ public class AttributeStores {
                 } else {
                     throw new RepositoryException(String.format("Unknown datatype %s", i.dataType()));
                 }
+            case ENUM:
+                return new IntAttributeStore(i);
             case ARRAY:
                 return new ImmutableListStore(i);
             case MAP:

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

@@ -42,6 +42,7 @@ public class DataTypes {
 
     public static enum TypeCategory {
         PRIMITIVE,
+        ENUM,
         ARRAY,
         MAP,
         STRUCT,

typesystem/src/main/java/org/apache/hadoop/metadata/types/EnumType.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.hadoop.metadata.types;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.hadoop.metadata.MetadataException;
+
+public class EnumType extends AbstractDataType<EnumValue> {
+
+    public final TypeSystem typeSystem;
+    public final String name;
+    public final ImmutableMap<String, EnumValue> valueMap;
+    public final ImmutableMap<Integer, EnumValue> ordinalMap;
+
+    protected EnumType(TypeSystem typeSystem, String name, EnumValue...values) {
+        this.typeSystem = typeSystem;
+        this.name = name;
+        ImmutableMap.Builder<String, EnumValue> b1 = new ImmutableMap.Builder();
+        ImmutableMap.Builder<Integer, EnumValue> b2 = new ImmutableMap.Builder();
+        for(EnumValue v : values) {
+            b1.put(v.value, v);
+            b2.put(v.ordinal, v);
+        }
+        valueMap = b1.build();
+        ordinalMap = b2.build();
+    }
+
+
+    @Override
+    public String getName() {
+        return name;
+    }
+
+    @Override
+    public EnumValue convert(Object val, Multiplicity m) throws MetadataException {
+        if ( val != null ) {
+            EnumValue e = null;
+            if ( val instanceof Integer) {
+                e = ordinalMap.get((Integer)val);
+            } else if ( val instanceof  String) {
+                e = valueMap.get((String)val);
+            } else if ( val instanceof Number ) {
+                e = ordinalMap.get(((Number)val).intValue());
+            }
+
+            if ( e == null ) {
+                throw new ValueConversionException(this, val);
+            }
+            return e;
+        }
+        return convertNull(m);
+    }
+
+    @Override
+    public DataTypes.TypeCategory getTypeCategory() {
+        return DataTypes.TypeCategory.ENUM;
+    }
+
+    public EnumValue fromOrdinal(int o) {
+        return ordinalMap.get(o);
+    }
+
+    public EnumValue fromValue(String val) {
+        return valueMap.get(val);
+    }
+}

typesystem/src/main/java/org/apache/hadoop/metadata/types/EnumValue.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.hadoop.metadata.types;
+
+public class EnumValue {
+
+    public final String value;
+    public final int ordinal;
+
+    public EnumValue(String value, int ordinal) {
+        this.value = value;
+        this.ordinal = ordinal;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        EnumValue enumValue = (EnumValue) o;
+
+        if (ordinal != enumValue.ordinal) return false;
+        if (!value.equals(enumValue.value)) return false;
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = value.hashCode();
+        result = 31 * result + ordinal;
+        return result;
+    }
+
+    @Override
+    public String toString() {
+        return value;
+    }
+}

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

@@ -221,6 +221,9 @@ public abstract class HierarchicalType<ST extends HierarchicalType,T> extends Ab
                 } else if ( i.dataType() == DataTypes.STRING_TYPE ) {
                     fieldPos.put(attrName, numStrings);
                     numStrings++;
+                } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.ENUM ) {
+                    fieldPos.put(i.name, numInts);
+                    numInts++;
                 } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.ARRAY ) {
                     fieldPos.put(attrName, numArrays);
                     numArrays++;

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

@@ -136,6 +136,9 @@ public class StructType  extends AbstractDataType<IStruct>
             } else if ( i.dataType() == DataTypes.STRING_TYPE ) {
                 fieldPos.put(i.name, numStrings);
                 numStrings++;
+            } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.ENUM ) {
+                fieldPos.put(i.name, numInts);
+                numInts++;
             } else if ( i.dataType().getTypeCategory() == DataTypes.TypeCategory.ARRAY ) {
                 fieldPos.put(i.name, numArrays);
                 numArrays++;

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

@@ -106,7 +106,7 @@ public class TypeSystem {
     ) throws MetadataException {
         Map<String, IDataType> newTypes = defineTypes(ImmutableList.<StructTypeDefinition>of(),
                 ImmutableList.<HierarchicalTypeDefinition<TraitType>>of(),
-                ImmutableList.<HierarchicalTypeDefinition<ClassType>>of());
+                ImmutableList.<HierarchicalTypeDefinition<ClassType>>of(classDef));
 
         return getDataType(ClassType.class, classDef.typeName);
     }
@@ -144,6 +144,16 @@ public class TypeSystem {
         return dT;
     }
 
+    public EnumType defineEnumType(String name, EnumValue...values) throws MetadataException {
+        assert name != null;
+        if (types.containsKey(name)) {
+            throw new MetadataException(String.format("Redefinition of type %s not supported", name));
+        }
+        EnumType eT = new EnumType(this, name, values);
+        types.put(name, eT);
+        return eT;
+    }
+
     class TransientTypeSystem extends TypeSystem {
 
         final ImmutableList<StructTypeDefinition> structDefs;

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

@@ -209,6 +209,10 @@ object Serialization {
         v = v.asInstanceOf[IReferenceableInstance].getId
       }
 
+      if ( v != null && (info.dataType().getTypeCategory eq TypeCategory.ENUM)) {
+        v = v.asInstanceOf[EnumValue].value
+      }
+
       JField(fName, Extraction.decompose(v))
     }
   }.toList.map(_.asInstanceOf[JField])