diff --git a/asterix-app/src/test/resources/runtimets/queries/types/type_promotion_0/type_promotion_0.1.ddl.aql b/asterix-app/src/test/resources/runtimets/queries/types/type_promotion_0/type_promotion_0.1.ddl.aql
new file mode 100644
index 0000000..c74fd3c
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/types/type_promotion_0/type_promotion_0.1.ddl.aql
@@ -0,0 +1,19 @@
+drop dataverse TestVerse if exists;
+create dataverse TestVerse;
+use dataverse TestVerse;
+
+create type Int64TestType as open {
+        myint64: int64,
+        myoptint64: int64?,
+        myint32: int32,
+        myoptint32: int32?,
+        myint16: int16,
+        myoptint16: int16?,
+        mydouble: double,
+        myoptdouble: double?,
+        myfloat: float,
+        myoptfloat: float?
+};
+
+create dataset Int64Test(Int64TestType)
+   primary key myint64;
diff --git a/asterix-app/src/test/resources/runtimets/queries/types/type_promotion_0/type_promotion_0.2.update.aql b/asterix-app/src/test/resources/runtimets/queries/types/type_promotion_0/type_promotion_0.2.update.aql
new file mode 100644
index 0000000..8505bff
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/types/type_promotion_0/type_promotion_0.2.update.aql
@@ -0,0 +1,13 @@
+use dataverse TestVerse;
+
+/* promotable type for optional field */
+insert into dataset Int64Test (
+       {"myint64": int64("13"), "myoptint64": 13, "myint32": int8("2"), "myoptint32": int16("3"), "myint16": int8("9"), "myoptint16": int8("10"), "mydouble": float("2.12"), "myoptdouble": int64("32"), "myfloat": int8("9"), "myoptfloat": int32("328")}
+);
+/* promotable type for non-optional field */
+insert into dataset Int64Test (
+       {"myint64": 12, "myoptint64": null, "myint32": int8("2"), "myoptint32": date(null), "myint16": int8("9"), "myoptint16": interval-starts(null, null), "mydouble": float("2.12"), "myoptdouble": time(null), "myfloat": int8("9"), "myoptfloat": datetime(null) }
+);
+insert into dataset Int64Test (
+       {"myint64": int16("11"), "myoptint64": int8("3"), "myint32": int8("2"), "myoptint32": int16("3"), "myint16": int8("9"), "myoptint16": int8("10"), "mydouble": int8("2"), "myoptdouble": int16("32"), "myfloat": int16("9"), "myoptfloat": datetime(null) }
+);
\ No newline at end of file
diff --git a/asterix-app/src/test/resources/runtimets/queries/types/type_promotion_0/type_promotion_0.3.query.aql b/asterix-app/src/test/resources/runtimets/queries/types/type_promotion_0/type_promotion_0.3.query.aql
new file mode 100644
index 0000000..8948b49
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/types/type_promotion_0/type_promotion_0.3.query.aql
@@ -0,0 +1,4 @@
+use dataverse TestVerse;
+
+for $i in dataset Int64Test
+return $i
\ No newline at end of file
diff --git a/asterix-app/src/test/resources/runtimets/results/types/type_promotion_0/type_promotion_0.1.adm b/asterix-app/src/test/resources/runtimets/results/types/type_promotion_0/type_promotion_0.1.adm
new file mode 100644
index 0000000..3c356ed
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/results/types/type_promotion_0/type_promotion_0.1.adm
@@ -0,0 +1,3 @@
+{ "myint64": 11i64, "myoptint64": 3i64, "myint32": 2, "myoptint32": 3, "myint16": 9i16, "myoptint16": 10i16, "mydouble": 2.0d, "myoptdouble": 32.0d, "myfloat": 9.0f, "myoptfloat": null }
+{ "myint64": 12i64, "myoptint64": null, "myint32": 2, "myoptint32": null, "myint16": 9i16, "myoptint16": null, "mydouble": 2.119999885559082d, "myoptdouble": null, "myfloat": 9.0f, "myoptfloat": null }
+{ "myint64": 13i64, "myoptint64": 13i64, "myint32": 2, "myoptint32": 3, "myint16": 9i16, "myoptint16": 10i16, "mydouble": 2.119999885559082d, "myoptdouble": 32.0d, "myfloat": 9.0f, "myoptfloat": 328.0f }
\ No newline at end of file
diff --git a/asterix-app/src/test/resources/runtimets/testsuite.xml b/asterix-app/src/test/resources/runtimets/testsuite.xml
index d36fbd4..a05806b 100644
--- a/asterix-app/src/test/resources/runtimets/testsuite.xml
+++ b/asterix-app/src/test/resources/runtimets/testsuite.xml
@@ -4568,5 +4568,10 @@
         <output-dir compare="Text">record01</output-dir>
       </compilation-unit>
     </test-case>
+    <test-case FilePath="types">
+      <compilation-unit name="type_promotion_0">
+        <output-dir compare="Text">type_promotion_0</output-dir>
+      </compilation-unit>
+    </test-case>
   </test-group>
 </test-suite>
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/FileSystemBasedAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/FileSystemBasedAdapter.java
index 33ee11f..753f7d1 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/FileSystemBasedAdapter.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/FileSystemBasedAdapter.java
@@ -16,7 +16,6 @@
 
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.List;
 import java.util.Map;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -29,6 +28,7 @@
 import edu.uci.ics.asterix.om.types.ATypeTag;
 import edu.uci.ics.asterix.om.types.AUnionType;
 import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
 import edu.uci.ics.asterix.runtime.operators.file.AdmSchemafullRecordParserFactory;
 import edu.uci.ics.asterix.runtime.operators.file.NtDelimitedDataTupleParserFactory;
 import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
@@ -108,11 +108,11 @@
         for (int i = 0; i < n; i++) {
             ATypeTag tag = null;
             if (recordType.getFieldTypes()[i].getTypeTag() == ATypeTag.UNION) {
-                List<IAType> unionTypes = ((AUnionType) recordType.getFieldTypes()[i]).getUnionList();
-                if (unionTypes.size() != 2 && unionTypes.get(0).getTypeTag() != ATypeTag.NULL) {
+                if (!NonTaggedFormatUtil.isOptionalField(((AUnionType) recordType.getFieldTypes()[i]))) {
                     throw new NotImplementedException("Non-optional UNION type is not supported.");
                 }
-                tag = unionTypes.get(1).getTypeTag();
+                tag = ((AUnionType) recordType.getFieldTypes()[i]).getUnionList()
+                        .get(NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST).getTypeTag();
             } else {
                 tag = recordType.getFieldTypes()[i].getTypeTag();
             }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ACastVisitor.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ACastVisitor.java
index 4e939ce..a711eb9 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ACastVisitor.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ACastVisitor.java
@@ -15,6 +15,7 @@
 
 package edu.uci.ics.asterix.om.pointables.cast;
 
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -28,8 +29,12 @@
 import edu.uci.ics.asterix.om.types.ARecordType;
 import edu.uci.ics.asterix.om.types.ATypeTag;
 import edu.uci.ics.asterix.om.types.AbstractCollectionType;
+import edu.uci.ics.asterix.om.types.EnumDeserializer;
 import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.types.hierachy.ATypeHierarchy;
+import edu.uci.ics.asterix.om.types.hierachy.ITypePromoteComputer;
 import edu.uci.ics.hyracks.algebricks.common.utils.Triple;
+import edu.uci.ics.hyracks.data.std.util.ArrayBackedValueStorage;
 
 /**
  * This class is a IVisitablePointableVisitor implementation which recursively
@@ -86,10 +91,48 @@
     }
 
     @Override
-    public Void visit(AFlatValuePointable accessor, Triple<IVisitablePointable, IAType, Boolean> arg) {
+    public Void visit(AFlatValuePointable accessor, Triple<IVisitablePointable, IAType, Boolean> arg)
+            throws AsterixException {
+        if (arg.second == null) {
+            // for open type case
+            arg.first.set(accessor);
+            return null;
+        }
         // set the pointer for result
-        arg.first.set(accessor);
+        ATypeTag reqTypeTag = ((IAType) (arg.second)).getTypeTag();
+        ATypeTag inputTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(accessor.getByteArray()[accessor
+                .getStartOffset()]);
+        if (!needPromote(inputTypeTag, reqTypeTag)) {
+            arg.first.set(accessor);
+        } else {
+            ArrayBackedValueStorage castBuffer = new ArrayBackedValueStorage();
+            ITypePromoteComputer promoteComputer = ATypeHierarchy.getTypePromoteComputer(inputTypeTag, reqTypeTag);
+            if (promoteComputer != null) {
+
+                try {
+                    // do the promotion; note that the type tag field should be skipped
+                    promoteComputer.promote(accessor.getByteArray(), accessor.getStartOffset() + 1,
+                            accessor.getLength() - 1, castBuffer);
+                    arg.first.set(castBuffer);
+                } catch (IOException e) {
+                    throw new AsterixException(e);
+                }
+            } else {
+                throw new AsterixException("Type mismatch: cannot cast type " + inputTypeTag + " to " + reqTypeTag);
+            }
+        }
+
         return null;
     }
 
+    private boolean needPromote(ATypeTag tag0, ATypeTag tag1) {
+        if (tag0 == tag1) {
+            return false;
+        }
+        if (tag0 == ATypeTag.NULL) {
+            return false;
+        }
+        return true;
+    }
+
 }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ARecordCaster.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ARecordCaster.java
index 507b845..2c4702c 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ARecordCaster.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ARecordCaster.java
@@ -36,6 +36,8 @@
 import edu.uci.ics.asterix.om.types.AUnionType;
 import edu.uci.ics.asterix.om.types.EnumDeserializer;
 import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.types.hierachy.ATypeHierarchy;
+import edu.uci.ics.asterix.om.types.hierachy.ITypePromoteComputer;
 import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
 import edu.uci.ics.asterix.om.util.ResettableByteArrayOutputStream;
 import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
@@ -88,6 +90,9 @@
     private int[] fieldNamesSortedIndex;
     private int[] reqFieldNamesSortedIndex;
 
+    // for promotion
+    private ITypePromoteComputer[] promoteComputers;
+
     public ARecordCaster() {
         try {
             bos.reset();
@@ -115,6 +120,7 @@
         if (openFields == null || numInputFields > openFields.length) {
             openFields = new boolean[numInputFields];
             fieldNamesSortedIndex = new int[numInputFields];
+            promoteComputers = new ITypePromoteComputer[numInputFields];
         }
         if (cachedReqType == null || !reqType.equals(cachedReqType)) {
             loadRequiredType(reqType);
@@ -134,6 +140,9 @@
             fieldPermutation[i] = -1;
         for (int i = 0; i < numInputFields; i++)
             fieldNamesSortedIndex[i] = i;
+        for (int i = 0; i < numInputFields; i++) {
+            promoteComputers[i] = null;
+        }
         outputBos.reset();
     }
 
@@ -205,6 +214,17 @@
                         optionalFields[reqFnPos] && fieldTypeTag.equals(nullTypeTag))) {
                     fieldPermutation[reqFnPos] = fnPos;
                     openFields[fnPos] = false;
+                } else {
+                    // try to do type promotion
+                    ATypeTag inputTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(fieldTypeTag
+                            .getByteArray()[fieldTypeTag.getStartOffset()]);
+                    ATypeTag requiredTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(reqFieldTypeTag
+                            .getByteArray()[reqFieldTypeTag.getStartOffset()]);
+
+                    if (ATypeHierarchy.canPromote(inputTypeTag, requiredTypeTag)) {
+                        fieldPermutation[reqFnPos] = fnPos;
+                        openFields[fnPos] = false;
+                    }
                 }
                 fnStart++;
                 reqFnStart++;
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/ARecordType.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/ARecordType.java
index 0319c7c..28a41d1 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/ARecordType.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/types/ARecordType.java
@@ -66,6 +66,7 @@
      * @throws AsterixException
      *             if there are duplicate field names or if there is an error serializing the field names
      */
+    @SuppressWarnings("resource")
     public ARecordType(String typeName, String[] fieldNames, IAType[] fieldTypes, boolean isOpen)
             throws AsterixException {
         super(typeName);
@@ -96,6 +97,11 @@
             hashCodeIndexPairs[i] = hashCodeIndexPairs[i] << 32;
             hashCodeIndexPairs[i] = hashCodeIndexPairs[i] | i;
         }
+        try {
+            dos.close();
+        } catch (IOException e) {
+            throw new AsterixException(e);
+        }
         serializedFieldNames = baaos.getByteArray();
 
         Arrays.sort(hashCodeIndexPairs);
@@ -336,7 +342,8 @@
                             break;
                         default:
                             throw new AlgebricksException("The field \"" + fieldName + "\" which is of type "
-                                    + fieldType.getTypeTag() + " cannot be indexed using the Length Partitioned N-Gram index.");
+                                    + fieldType.getTypeTag()
+                                    + " cannot be indexed using the Length Partitioned N-Gram index.");
                     }
                     break;
                 case LENGTH_PARTITIONED_WORD_INVIX:
@@ -348,7 +355,8 @@
                             break;
                         default:
                             throw new AlgebricksException("The field \"" + fieldName + "\" which is of type "
-                                    + fieldType.getTypeTag() + " cannot be indexed using the Length Partitioned Keyword index.");
+                                    + fieldType.getTypeTag()
+                                    + " cannot be indexed using the Length Partitioned Keyword index.");
                     }
                     break;
                 case SINGLE_PARTITION_NGRAM_INVIX:
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/operators/file/DelimitedDataParser.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/operators/file/DelimitedDataParser.java
index 5a639dc..23f7aab 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/operators/file/DelimitedDataParser.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/operators/file/DelimitedDataParser.java
@@ -29,6 +29,8 @@
 import edu.uci.ics.asterix.om.base.ANull;
 import edu.uci.ics.asterix.om.types.ARecordType;
 import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.AUnionType;
+import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.data.std.util.ArrayBackedValueStorage;
 import edu.uci.ics.hyracks.dataflow.common.data.parsers.IValueParser;
@@ -115,8 +117,9 @@
                         && recordType.getFieldTypes()[i].getTypeTag() != ATypeTag.NULL) {
                     // if the field is empty and the type is optional, insert NULL
                     // note that string type can also process empty field as an empty string
-                    if (recordType.getFieldTypes()[i].getTypeTag() != ATypeTag.UNION) {
-                        throw new AsterixException("Field " + i + " cannot be NULL. ");
+                    if (recordType.getFieldTypes()[i].getTypeTag() != ATypeTag.UNION
+                            || !NonTaggedFormatUtil.isOptionalField((AUnionType) recordType.getFieldTypes()[i])) {
+                        throw new AsterixException("Field " + i + " is not an optional type so it cannot accept null value. ");
                     }
                     fieldValueBufferOutput.writeByte(ATypeTag.NULL.serialize());
                     ANullSerializerDeserializer.INSTANCE.serialize(ANull.NULL, out);