add primitive attribute stores

typesystem/src/main/java/org/apache/hadoop/metadata/ITypedInstance.java

@@ -18,8 +18,14 @@
 
 package org.apache.hadoop.metadata;
 
+import org.apache.hadoop.metadata.types.AttributeInfo;
+import org.apache.hadoop.metadata.types.DataTypes;
 import org.apache.hadoop.metadata.types.FieldMapping;
 
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.Date;
+
 /**
  * An instance whose structure is associated with a IDataType.
  * This is obtained by a call to 'createInstance' or the result of a Query.
@@ -31,4 +37,30 @@ import org.apache.hadoop.metadata.types.FieldMapping;
 public interface ITypedInstance extends IInstance {
 
     FieldMapping fieldMapping();
+
+    public void setNull(String attrName) throws MetadataException;
+
+    public boolean getBoolean(String attrName) throws MetadataException;
+    public byte getByte(String attrName) throws MetadataException;
+    public short getShort(String attrName) throws MetadataException;
+    public int getInt(String attrName) throws MetadataException;
+    public long getLong(String attrName) throws MetadataException;
+    public float getFloat(String attrName) throws MetadataException;
+    public double getDouble(String attrName) throws MetadataException;
+    public BigInteger getBigInt(String attrName) throws MetadataException;
+    public BigDecimal getBigDecimal(String attrName) throws MetadataException ;
+    public Date getDate(String attrName) throws MetadataException;
+    public String getString(String attrName) throws MetadataException;
+
+    public void setBoolean(String attrName, boolean val) throws MetadataException;
+    public void setByte(String attrName, byte val) throws MetadataException;
+    public void setShort(String attrName, short val) throws MetadataException;
+    public void setInt(String attrName, int val) throws MetadataException;
+    public void setLong(String attrName, long val) throws MetadataException;
+    public void setFloat(String attrName, float val) throws MetadataException;
+    public void setDouble(String attrName, double val) throws MetadataException;
+    public void setBigInt(String attrName, BigInteger val) throws MetadataException;
+    public void setBigDecimal(String attrName, BigDecimal val) throws MetadataException;
+    public void setDate(String attrName, Date val) throws MetadataException;
+    public void setString(String attrName, String val) throws MetadataException;
 }

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

@@ -25,6 +25,10 @@ import org.apache.hadoop.metadata.ITypedReferenceableInstance;
 import org.apache.hadoop.metadata.MetadataException;
 import org.apache.hadoop.metadata.types.FieldMapping;
 
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.Date;
+
 public class Id implements ITypedReferenceableInstance {
 
     public final long id;
@@ -38,7 +42,7 @@ public class Id implements ITypedReferenceableInstance {
     }
 
     public Id(String className) {
-        this(- System.currentTimeMillis(), 0, className);
+        this(-System.currentTimeMillis(), 0, className);
     }
 
     public boolean isUnassigned() {
@@ -83,4 +87,96 @@ public class Id implements ITypedReferenceableInstance {
     public FieldMapping fieldMapping() {
         return null;
     }
+
+    public void setNull(String attrName) throws MetadataException {
+        set(attrName, null);
+    }
+
+    public boolean getBoolean(String attrName) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public byte getByte(String attrName) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public short getShort(String attrName) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public int getInt(String attrName) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public long getLong(String attrName) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public float getFloat(String attrName) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public double getDouble(String attrName) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public BigInteger getBigInt(String attrName) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public BigDecimal getBigDecimal(String attrName) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public Date getDate(String attrName) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public String getString(String attrName) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public void setBoolean(String attrName, boolean val) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public void setByte(String attrName, byte val) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public void setShort(String attrName, short val) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public void setInt(String attrName, int val) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public void setLong(String attrName, long val) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public void setFloat(String attrName, float val) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public void setDouble(String attrName, double val) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public void setBigInt(String attrName, BigInteger val) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public void setBigDecimal(String attrName, BigDecimal val) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public void setDate(String attrName, Date val) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
+
+    public void setString(String attrName, String val) throws MetadataException {
+        throw new MetadataException("Get/Set not supported on an Id object");
+    }
 }

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

@@ -208,6 +208,443 @@ public class StructInstance implements ITypedStruct {
         }
     }
 
+    public void setNull(String attrName) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+        nullFlags[nullPos] = true;
+    }
+
+    public boolean getBoolean(String attrName) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.BOOLEAN_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic get method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        if ( nullFlags[nullPos]) {
+            return DataTypes.BOOLEAN_TYPE.nullValue();
+        }
+
+        return bools[pos];
+    }
+
+    public byte getByte(String attrName) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.BYTE_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic get method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        if ( nullFlags[nullPos]) {
+            return DataTypes.BYTE_TYPE.nullValue();
+        }
+
+        return bytes[pos];
+    }
+
+    public short getShort(String attrName) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.SHORT_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic get method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        if ( nullFlags[nullPos]) {
+            return DataTypes.SHORT_TYPE.nullValue();
+        }
+
+        return shorts[pos];
+    }
+
+    public int getInt(String attrName) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.INT_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic get method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        if ( nullFlags[nullPos]) {
+            return DataTypes.INT_TYPE.nullValue();
+        }
+
+        return ints[pos];
+    }
+
+    public long getLong(String attrName) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.LONG_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic get method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        if ( nullFlags[nullPos]) {
+            return DataTypes.LONG_TYPE.nullValue();
+        }
+
+        return longs[pos];
+    }
+
+    public float getFloat(String attrName) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.FLOAT_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic get method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        if ( nullFlags[nullPos]) {
+            return DataTypes.FLOAT_TYPE.nullValue();
+        }
+
+        return floats[pos];
+    }
+
+    public double getDouble(String attrName) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.DOUBLE_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic get method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        if ( nullFlags[nullPos]) {
+            return DataTypes.DOUBLE_TYPE.nullValue();
+        }
+
+        return doubles[pos];
+    }
+
+    public BigInteger getBigInt(String attrName) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.BIGINTEGER_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic get method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        if ( nullFlags[nullPos]) {
+            return DataTypes.BIGINTEGER_TYPE.nullValue();
+        }
+
+        return bigIntegers[pos];
+    }
+
+    public BigDecimal getBigDecimal(String attrName) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.BIGDECIMAL_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic get method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        if ( nullFlags[nullPos]) {
+            return DataTypes.BIGDECIMAL_TYPE.nullValue();
+        }
+
+        return bigDecimals[pos];
+    }
+
+    public Date getDate(String attrName) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.DATE_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic get method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        if ( nullFlags[nullPos]) {
+            return DataTypes.DATE_TYPE.nullValue();
+        }
+
+        return dates[pos];
+    }
+
+    public String getString(String attrName) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.STRING_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic get method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        if ( nullFlags[nullPos]) {
+            return DataTypes.STRING_TYPE.nullValue();
+        }
+
+        return strings[pos];
+    }
+
+    public void setBoolean(String attrName, boolean val) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.BOOLEAN_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic set method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        nullFlags[nullPos] = false;
+        bools[pos] = val;
+    }
+
+    public void setByte(String attrName, byte val) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.BYTE_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic set method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        nullFlags[nullPos] = false;
+        bytes[pos] = val;
+    }
+
+    public void setShort(String attrName, short val) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.SHORT_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic set method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        nullFlags[nullPos] = false;
+        shorts[pos] = val;
+    }
+
+    public void setInt(String attrName, int val) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.INT_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic set method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        nullFlags[nullPos] = false;
+        ints[pos] = val;
+    }
+
+    public void setLong(String attrName, long val) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.LONG_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic set method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        nullFlags[nullPos] = false;
+        longs[pos] = val;
+    }
+
+    public void setFloat(String attrName, float val) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.FLOAT_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic set method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        nullFlags[nullPos] = false;
+        floats[pos] = val;
+    }
+
+    public void setDouble(String attrName, double val) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if (i == null) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if (i.dataType() != DataTypes.DOUBLE_TYPE) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic set method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        nullFlags[nullPos] = false;
+        doubles[pos] = val;
+    }
+
+    public void setBigInt(String attrName, BigInteger val) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.BIGINTEGER_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic set method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        nullFlags[nullPos] = val == null;
+        bigIntegers[pos] = val;
+    }
+
+    public void setBigDecimal(String attrName, BigDecimal val) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.BIGDECIMAL_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic set method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        nullFlags[nullPos] = val == null;
+        bigDecimals[pos] = val;
+    }
+
+    public void setDate(String attrName, Date val) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.DATE_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic set method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        nullFlags[nullPos] = val == null;
+        dates[pos] = val;
+    }
+
+    public void setString(String attrName, String val) throws MetadataException {
+        AttributeInfo i = fieldMapping.fields.get(attrName);
+        if ( i == null ) {
+            throw new MetadataException(String.format("Unknown field %s for Struct %s", attrName, getTypeName()));
+        }
+
+        if ( i.dataType() != DataTypes.STRING_TYPE ) {
+            throw new MetadataException(String.format("Field %s for Struct %s is not a %s, call generic set method",
+                    attrName, getTypeName()));
+        }
+
+        int pos = fieldMapping.fieldPos.get(attrName);
+        int nullPos = fieldMapping.fieldNullPos.get(attrName);
+
+        nullFlags[nullPos] = val == null;
+        strings[pos] = val;
+    }
 
     public void output(IStruct s, Appendable buf, String prefix) throws MetadataException {
         TypeUtils.outputVal("{", buf, prefix);

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

@@ -18,28 +18,65 @@
 
 package org.apache.hadoop.metadata.storage.memory;
 
+import com.google.common.collect.Lists;
 import it.unimi.dsi.fastutil.booleans.BooleanArrayList;
+import it.unimi.dsi.fastutil.bytes.ByteArrayList;
+import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import it.unimi.dsi.fastutil.floats.FloatArrayList;
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+import it.unimi.dsi.fastutil.longs.LongArrayList;
+import it.unimi.dsi.fastutil.shorts.ShortArrayList;
 import org.apache.hadoop.metadata.storage.StructInstance;
 import org.apache.hadoop.metadata.ITypedInstance;
 import org.apache.hadoop.metadata.storage.RepositoryException;
 import org.apache.hadoop.metadata.storage.StructInstance;
 import org.apache.hadoop.metadata.types.AttributeInfo;
+import org.apache.hadoop.metadata.types.DataTypes;
 import org.apache.hadoop.metadata.types.FieldMapping;
 import org.apache.hadoop.metadata.types.IConstructableType;
 
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.*;
 
 public class AttributeStores {
 
     private static final Object NULL_VAL = new Object();
 
-    static IAttributeStore createStore(AttributeInfo i) {
-        return null;
+    static IAttributeStore createStore(AttributeInfo i) throws RepositoryException {
+        switch(i.dataType().getTypeCategory() ) {
+            case PRIMITIVE:
+                if ( i.dataType() == DataTypes.BOOLEAN_TYPE ) {
+                    return new BooleanAttributeStore(i);
+                } else if ( i.dataType() == DataTypes.BYTE_TYPE ) {
+                    return new ByteAttributeStore(i);
+                } else if ( i.dataType() == DataTypes.SHORT_TYPE ) {
+                    new ShortAttributeStore(i);
+                } else if ( i.dataType() == DataTypes.INT_TYPE ) {
+                    new IntAttributeStore(i);
+                } else if ( i.dataType() == DataTypes.LONG_TYPE ) {
+                    new LongAttributeStore(i);
+                } else if ( i.dataType() == DataTypes.FLOAT_TYPE ) {
+                    new FloatAttributeStore(i);
+                } else if ( i.dataType() == DataTypes.DOUBLE_TYPE ) {
+                    new DoubleAttributeStore(i);
+                } else if ( i.dataType() == DataTypes.BIGINTEGER_TYPE ) {
+                    new BigIntStore(i);
+                } else if ( i.dataType() == DataTypes.BIGDECIMAL_TYPE ) {
+                    new BigDecimalStore(i);
+                } else if ( i.dataType() == DataTypes.DATE_TYPE ) {
+                    new DateStore(i);
+                } else if ( i.dataType() == DataTypes.STRING_TYPE ) {
+                    new StringStore(i);
+                } else {
+                    throw new RepositoryException(String.format("Unknown datatype %s", i.dataType()));
+                }
+            default:
+                throw new RepositoryException(String.format("Unknown Category for datatype %s", i.dataType()));
+        }
     }
 
-    static abstract class AbstractAttributeStore {
+    static abstract class AbstractAttributeStore implements IAttributeStore {
         AttributeInfo attrInfo;
         final BooleanArrayList nullList;
         final Map<Integer, Map<String, Object>> hiddenVals;
@@ -61,30 +98,37 @@ public class AttributeStores {
         void storeHiddenVals(int pos, IConstructableType type, StructInstance instance) throws RepositoryException {
             List<String> attrNames = type.getNames(attrInfo);
             Map<String, Object> m = hiddenVals.get(pos);
-            if ( m == null ) {
+            if (m == null) {
                 m = new HashMap<String, Object>();
                 hiddenVals.put(pos, m);
             }
-            for(int i=2; i < attrNames.size(); i++ ) {
+            for (int i = 2; i < attrNames.size(); i++) {
                 String attrName = attrNames.get(i);
                 int nullPos = instance.fieldMapping().fieldNullPos.get(attrName);
                 int colPos = instance.fieldMapping().fieldPos.get(attrName);
-                if ( instance.nullFlags[nullPos] ) {
+                if (instance.nullFlags[nullPos]) {
                     m.put(attrName, NULL_VAL);
-                } else{
-                    m.put(attrName, instance.bools[colPos]);
+                } else {
+                    //m.put(attrName, instance.bools[colPos]);
+                    store(instance, colPos, attrName, m);
                 }
             }
         }
-    }
-
-    static class BooleanAttributeStore extends AbstractAttributeStore implements IAttributeStore {
 
-        final BooleanArrayList list;
-
-        BooleanAttributeStore(AttributeInfo attrInfo) {
-            super(attrInfo);
-            this.list = new BooleanArrayList();
+        void loadHiddenVals(int pos, IConstructableType type, StructInstance instance) throws RepositoryException {
+            List<String> attrNames = type.getNames(attrInfo);
+            Map<String, Object> m = hiddenVals.get(pos);
+            for (int i = 2; i < attrNames.size(); i++) {
+                String attrName = attrNames.get(i);
+                int nullPos = instance.fieldMapping().fieldNullPos.get(attrName);
+                int colPos = instance.fieldMapping().fieldPos.get(attrName);
+                Object val = m == null ? NULL_VAL : m.get(attrName);
+                if (val == NULL_VAL) {
+                    instance.nullFlags[nullPos] = true;
+                } else {
+                    load(instance, colPos, val);
+                }
+            }
         }
 
         @Override
@@ -94,16 +138,85 @@ public class AttributeStores {
             int nullPos = instance.fieldMapping().fieldNullPos.get(attrName);
             int colPos = instance.fieldMapping().fieldPos.get(attrName);
             nullList.set(pos, instance.nullFlags[nullPos]);
-            list.set(pos, instance.bools[colPos]);
+            //list.set(pos, instance.bools[colPos]);
+            store(instance, colPos, pos);
 
-            if ( attrNames.size() > 1 ) {
+            if (attrNames.size() > 1) {
                 storeHiddenVals(pos, type, instance);
             }
         }
 
         @Override
-        public void load(int pos, IConstructableType type,  StructInstance instance) throws RepositoryException {
+        public void load(int pos, IConstructableType type, StructInstance instance) throws RepositoryException {
+            List<String> attrNames = type.getNames(attrInfo);
+            String attrName = attrNames.get(0);
+            int nullPos = instance.fieldMapping().fieldNullPos.get(attrName);
+            int colPos = instance.fieldMapping().fieldPos.get(attrName);
+
+            if (nullList.get(pos)) {
+                instance.nullFlags[colPos] = true;
+            } else {
+                load(instance, colPos, pos);
+            }
+
+            if (attrNames.size() > 1) {
+                loadHiddenVals(pos, type, instance);
+            }
+        }
+
+        /*
+         * store the value from colPos in instance into the list.
+         */
+        protected abstract void store(StructInstance instance, int colPos, int pos);
+
+        /*
+         * load the value from pos in list into colPos in instance.
+         */
+        protected abstract void load(StructInstance instance, int colPos, int pos);
+        /*
+         * store the value from colPos in map as attrName
+         */
+        protected abstract void store(StructInstance instance, int colPos, String attrName, Map<String, Object> m);
+
+        /*
+         * load the val into colPos in instance.
+         */
+        protected abstract void load(StructInstance instance, int colPos, Object val);
+
+    }
+
+    static abstract class PrimitiveAttributeStore extends AbstractAttributeStore implements IAttributeStore {
+
+
+        public PrimitiveAttributeStore(AttributeInfo attrInfo) {
+            super(attrInfo);
+        }
+
+    }
+
+    static class BooleanAttributeStore extends PrimitiveAttributeStore {
+
+        final BooleanArrayList list;
+
+        BooleanAttributeStore(AttributeInfo attrInfo) {
+            super(attrInfo);
+            this.list = new BooleanArrayList();
+        }
+
+        protected void store(StructInstance instance, int colPos, int pos) {
+            list.set(pos, instance.bools[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, int pos) {
+            instance.bools[colPos] = list.get(pos);
+        }
+
+        protected void store(StructInstance instance, int colPos, String attrName, Map<String, Object> m) {
+            m.put(attrName, instance.bools[colPos]);
+        }
 
+        protected void load(StructInstance instance, int colPos, Object val) {
+            instance.bools[colPos] = (Boolean) val;
         }
 
         @Override
@@ -112,4 +225,308 @@ public class AttributeStores {
             nullList.ensureCapacity(pos);
         }
     }
+
+    static class ByteAttributeStore extends PrimitiveAttributeStore {
+
+        final ByteArrayList list;
+
+        ByteAttributeStore(AttributeInfo attrInfo) {
+            super(attrInfo);
+            this.list = new ByteArrayList();
+        }
+
+        protected void store(StructInstance instance, int colPos, int pos) {
+            list.set(pos, instance.bytes[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, int pos) {
+            instance.bytes[colPos] = list.get(pos);
+        }
+
+        protected void store(StructInstance instance, int colPos, String attrName, Map<String, Object> m) {
+            m.put(attrName, instance.bytes[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, Object val) {
+            instance.bytes[colPos] = (Byte) val;
+        }
+
+        @Override
+        public void ensureCapacity(int pos) throws RepositoryException {
+            list.ensureCapacity(pos);
+            nullList.ensureCapacity(pos);
+        }
+    }
+
+    static class ShortAttributeStore extends PrimitiveAttributeStore {
+
+        final ShortArrayList list;
+
+        ShortAttributeStore(AttributeInfo attrInfo) {
+            super(attrInfo);
+            this.list = new ShortArrayList();
+        }
+
+        protected void store(StructInstance instance, int colPos, int pos) {
+            list.set(pos, instance.shorts[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, int pos) {
+            instance.shorts[colPos] = list.get(pos);
+        }
+
+        protected void store(StructInstance instance, int colPos, String attrName, Map<String, Object> m) {
+            m.put(attrName, instance.shorts[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, Object val) {
+            instance.shorts[colPos] = (Short) val;
+        }
+
+        @Override
+        public void ensureCapacity(int pos) throws RepositoryException {
+            list.ensureCapacity(pos);
+            nullList.ensureCapacity(pos);
+        }
+    }
+
+    static class IntAttributeStore extends PrimitiveAttributeStore {
+
+        final IntArrayList list;
+
+        IntAttributeStore(AttributeInfo attrInfo) {
+            super(attrInfo);
+            this.list = new IntArrayList();
+        }
+
+        protected void store(StructInstance instance, int colPos, int pos) {
+            list.set(pos, instance.ints[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, int pos) {
+            instance.ints[colPos] = list.get(pos);
+        }
+
+        protected void store(StructInstance instance, int colPos, String attrName, Map<String, Object> m) {
+            m.put(attrName, instance.ints[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, Object val) {
+            instance.ints[colPos] = (Integer) val;
+        }
+
+        @Override
+        public void ensureCapacity(int pos) throws RepositoryException {
+            list.ensureCapacity(pos);
+            nullList.ensureCapacity(pos);
+        }
+    }
+
+    static class LongAttributeStore extends PrimitiveAttributeStore {
+
+        final LongArrayList list;
+
+        LongAttributeStore(AttributeInfo attrInfo) {
+            super(attrInfo);
+            this.list = new LongArrayList();
+        }
+
+        protected void store(StructInstance instance, int colPos, int pos) {
+            list.set(pos, instance.longs[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, int pos) {
+            instance.longs[colPos] = list.get(pos);
+        }
+
+        protected void store(StructInstance instance, int colPos, String attrName, Map<String, Object> m) {
+            m.put(attrName, instance.longs[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, Object val) {
+            instance.longs[colPos] = (Long) val;
+        }
+
+        @Override
+        public void ensureCapacity(int pos) throws RepositoryException {
+            list.ensureCapacity(pos);
+            nullList.ensureCapacity(pos);
+        }
+    }
+
+    static class FloatAttributeStore extends PrimitiveAttributeStore {
+
+        final FloatArrayList list;
+
+        FloatAttributeStore(AttributeInfo attrInfo) {
+            super(attrInfo);
+            this.list = new FloatArrayList();
+        }
+
+        protected void store(StructInstance instance, int colPos, int pos) {
+            list.set(pos, instance.floats[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, int pos) {
+            instance.floats[colPos] = list.get(pos);
+        }
+
+        protected void store(StructInstance instance, int colPos, String attrName, Map<String, Object> m) {
+            m.put(attrName, instance.floats[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, Object val) {
+            instance.floats[colPos] = (Float) val;
+        }
+
+        @Override
+        public void ensureCapacity(int pos) throws RepositoryException {
+            list.ensureCapacity(pos);
+            nullList.ensureCapacity(pos);
+        }
+    }
+
+    static class DoubleAttributeStore extends PrimitiveAttributeStore {
+
+        final DoubleArrayList list;
+
+        DoubleAttributeStore(AttributeInfo attrInfo) {
+            super(attrInfo);
+            this.list = new DoubleArrayList();
+        }
+
+        protected void store(StructInstance instance, int colPos, int pos) {
+            list.set(pos, instance.doubles[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, int pos) {
+            instance.doubles[colPos] = list.get(pos);
+        }
+
+        protected void store(StructInstance instance, int colPos, String attrName, Map<String, Object> m) {
+            m.put(attrName, instance.doubles[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, Object val) {
+            instance.doubles[colPos] = (Double) val;
+        }
+
+        @Override
+        public void ensureCapacity(int pos) throws RepositoryException {
+            list.ensureCapacity(pos);
+            nullList.ensureCapacity(pos);
+        }
+    }
+
+    static abstract class ObjectAttributeStore<T> extends AbstractAttributeStore {
+
+        final ArrayList<T> list;
+
+        ObjectAttributeStore(Class<T> cls, AttributeInfo attrInfo) {
+            super(attrInfo);
+            this.list = Lists.newArrayList((T)null);
+        }
+
+        @Override
+        public void ensureCapacity(int pos) throws RepositoryException {
+            list.ensureCapacity(pos);
+            nullList.ensureCapacity(pos);
+        }
+    }
+
+    static class BigIntStore extends ObjectAttributeStore<BigInteger> {
+
+        public BigIntStore( AttributeInfo attrInfo) {
+            super(BigInteger.class, attrInfo);
+        }
+
+        protected void store(StructInstance instance, int colPos, int pos) {
+            list.set(pos, instance.bigIntegers[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, int pos) {
+            instance.bigIntegers[colPos] = list.get(pos);
+        }
+
+        protected void store(StructInstance instance, int colPos, String attrName, Map<String, Object> m) {
+            m.put(attrName, instance.bigIntegers[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, Object val) {
+            instance.bigIntegers[colPos] = (BigInteger) val;
+        }
+
+    }
+
+    static class BigDecimalStore extends ObjectAttributeStore<BigDecimal> {
+
+        public BigDecimalStore( AttributeInfo attrInfo) {
+            super(BigDecimal.class, attrInfo);
+        }
+
+        protected void store(StructInstance instance, int colPos, int pos) {
+            list.set(pos, instance.bigDecimals[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, int pos) {
+            instance.bigDecimals[colPos] = list.get(pos);
+        }
+
+        protected void store(StructInstance instance, int colPos, String attrName, Map<String, Object> m) {
+            m.put(attrName, instance.bigDecimals[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, Object val) {
+            instance.bigDecimals[colPos] = (BigDecimal) val;
+        }
+
+    }
+
+    static class DateStore extends ObjectAttributeStore<Date> {
+
+        public DateStore( AttributeInfo attrInfo) {
+            super(Date.class, attrInfo);
+        }
+
+        protected void store(StructInstance instance, int colPos, int pos) {
+            list.set(pos, instance.dates[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, int pos) {
+            instance.dates[colPos] = list.get(pos);
+        }
+
+        protected void store(StructInstance instance, int colPos, String attrName, Map<String, Object> m) {
+            m.put(attrName, instance.dates[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, Object val) {
+            instance.dates[colPos] = (Date) val;
+        }
+
+    }
+
+    static class StringStore extends ObjectAttributeStore<String> {
+
+        public StringStore( AttributeInfo attrInfo) {
+            super(String.class, attrInfo);
+        }
+
+        protected void store(StructInstance instance, int colPos, int pos) {
+            list.set(pos, instance.strings[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, int pos) {
+            instance.strings[colPos] = list.get(pos);
+        }
+
+        protected void store(StructInstance instance, int colPos, String attrName, Map<String, Object> m) {
+            m.put(attrName, instance.strings[colPos]);
+        }
+
+        protected void load(StructInstance instance, int colPos, Object val) {
+            instance.strings[colPos] = (String) val;
+        }
+
+    }
 }

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

@@ -37,7 +37,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
-public class StructStore extends AttributeStores.AbstractAttributeStore implements IAttributeStore {
+public abstract class StructStore extends AttributeStores.AbstractAttributeStore implements IAttributeStore {
 
     final StructType structType;
     final ImmutableMap<AttributeInfo, IAttributeStore> attrStores;

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

@@ -49,12 +49,14 @@ public class DataTypes {
         CLASS;
     };
 
-    static abstract class PrimitiveType<T> extends AbstractDataType<T> {
+    public static abstract class PrimitiveType<T> extends AbstractDataType<T> {
         @Override
         public TypeCategory getTypeCategory() {
             return TypeCategory.PRIMITIVE;
         }
 
+        public abstract T nullValue();
+
     }
 
     public static BooleanType BOOLEAN_TYPE = new BooleanType();
@@ -84,6 +86,10 @@ public class DataTypes {
             }
             return convertNull(m);
         }
+
+        public Boolean nullValue() {
+            return Boolean.FALSE;
+        }
     }
 
     public static ByteType BYTE_TYPE = new ByteType();
@@ -113,6 +119,10 @@ public class DataTypes {
             }
             return convertNull(m);
         }
+
+        public Byte nullValue() {
+            return 0;
+        }
     }
 
     public static ShortType SHORT_TYPE = new ShortType();
@@ -142,6 +152,10 @@ public class DataTypes {
             }
             return convertNull(m);
         }
+
+        public Short nullValue() {
+            return 0;
+        }
     }
 
     public static IntType INT_TYPE = new IntType();
@@ -171,6 +185,10 @@ public class DataTypes {
             }
             return convertNull(m);
         }
+
+        public Integer nullValue() {
+            return 0;
+        }
     }
 
     public static LongType LONG_TYPE = new LongType();
@@ -200,6 +218,10 @@ public class DataTypes {
             }
             return convertNull(m);
         }
+
+        public Long nullValue() {
+            return 0l;
+        }
     }
 
     public static FloatType FLOAT_TYPE = new FloatType();
@@ -229,6 +251,10 @@ public class DataTypes {
             }
             return convertNull(m);
         }
+
+        public Float nullValue() {
+            return 0.0f;
+        }
     }
 
     public static DoubleType DOUBLE_TYPE = new DoubleType();
@@ -258,6 +284,10 @@ public class DataTypes {
             }
             return convertNull(m);
         }
+
+        public Double nullValue() {
+            return 0.0;
+        }
     }
 
     public static BigIntegerType BIGINTEGER_TYPE = new BigIntegerType();
@@ -293,6 +323,10 @@ public class DataTypes {
             }
             return convertNull(m);
         }
+
+        public BigInteger nullValue() {
+            return null;
+        }
     }
 
     public static BigDecimalType BIGDECIMAL_TYPE = new BigDecimalType();
@@ -328,6 +362,10 @@ public class DataTypes {
             }
             return convertNull(m);
         }
+
+        public BigDecimal nullValue() {
+            return null;
+        }
     }
 
     public static DateType DATE_TYPE = new DateType();
@@ -361,6 +399,10 @@ public class DataTypes {
             }
             return convertNull(m);
         }
+
+        public Date nullValue() {
+            return null;
+        }
     }
 
     public static StringType STRING_TYPE = new StringType();
@@ -382,6 +424,10 @@ public class DataTypes {
             }
             return convertNull(m);
         }
+
+        public String nullValue() {
+            return null;
+        }
     }
 
     static String ARRAY_TYPE_PREFIX = "array<";
@@ -423,7 +469,7 @@ public class DataTypes {
                     it = (Iterator)val;
                 }
                 if ( it != null ) {
-                    ImmutableCollection.Builder<?> b = m.isUnique ? ImmutableSet.builder() : ImmutableList.builder();
+                    ImmutableCollection.Builder b = m.isUnique ? ImmutableSet.builder() : ImmutableList.builder();
                     while (it.hasNext() ) {
                         b.add(elemType.convert(it.next(),
                                 r.allowNullsInCollections() ? Multiplicity.OPTIONAL : Multiplicity.REQUIRED));
@@ -451,7 +497,7 @@ public class DataTypes {
             if ( val == null || elemType.getTypeCategory() != TypeCategory.CLASS ) {
                 return val;
             }
-            ImmutableCollection.Builder<?> b = m.isUnique ? ImmutableSet.builder() : ImmutableList.builder();
+            ImmutableCollection.Builder b = m.isUnique ? ImmutableSet.builder() : ImmutableList.builder();
             Iterator it = val.iterator();
             while(it.hasNext()) {
                 Object elem = it.next();