diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceDynamicTypeCastRule.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceDynamicTypeCastRule.java
index 1f242b8..4728969 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceDynamicTypeCastRule.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceDynamicTypeCastRule.java
@@ -26,7 +26,10 @@
 import edu.uci.ics.asterix.om.functions.AsterixBuiltinFunctions;
 import edu.uci.ics.asterix.om.typecomputer.base.TypeComputerUtilities;
 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.types.IAType;
+import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
@@ -40,7 +43,6 @@
 import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
 import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AssignOperator;
 import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.InsertDeleteOperator;
-import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.ProjectOperator;
 import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
 import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
 
@@ -88,62 +90,114 @@
         InsertDeleteOperator insertDeleteOp = (InsertDeleteOperator) op2;
         if (insertDeleteOp.getOperation() == InsertDeleteOperator.Kind.DELETE)
             return false;
-        AbstractLogicalOperator op3 = (AbstractLogicalOperator) op2.getInputs().get(0).getValue();
-        if (op3.getOperatorTag() != LogicalOperatorTag.ASSIGN)
-            return false;
 
         InsertDeleteOperator insertDeleteOperator = (InsertDeleteOperator) op2;
-        AssignOperator oldAssignOperator = (AssignOperator) op3;
-
         AqlDataSource dataSource = (AqlDataSource) insertDeleteOperator.getDataSource();
         IAType[] schemaTypes = (IAType[]) dataSource.getSchemaTypes();
         ARecordType requiredRecordType = (ARecordType) schemaTypes[schemaTypes.length - 1];
-
-        List<LogicalVariable> usedVariables = new ArrayList<LogicalVariable>();
-        VariableUtilities.getUsedVariables(oldAssignOperator, usedVariables);
-        LogicalVariable inputRecordVar;
-        if (usedVariables.size() > 0) {
-            inputRecordVar = usedVariables.get(0);
-        } else {
-            VariableUtilities.getLiveVariables(oldAssignOperator, usedVariables);
-            inputRecordVar = usedVariables.get(0);
-        }
-        IVariableTypeEnvironment env = oldAssignOperator.computeInputTypeEnvironment(context);
-        IAType inputRecordType = (IAType) env.getVarType(inputRecordVar);
-        boolean needCast = !requiredRecordType.equals(inputRecordType);
-        if (!needCast)
+        ILogicalExpression expr = insertDeleteOperator.getPayloadExpression().getValue();
+        List<LogicalVariable> payloadVars = new ArrayList<LogicalVariable>();
+        expr.getUsedVariables(payloadVars);
+        LogicalVariable recordVar = payloadVars.get(0);
+        IVariableTypeEnvironment env = insertDeleteOperator.computeOutputTypeEnvironment(context);
+        IAType inputRecordType = (IAType) env.getVarType(recordVar);
+        if (capatible(requiredRecordType, inputRecordType)) {
             return false;
+        }
 
-        // insert
-        // project
-        // assign
-        // assign
-        AbstractFunctionCallExpression cast = new ScalarFunctionCallExpression(
-                FunctionUtils.getFunctionInfo(AsterixBuiltinFunctions.CAST_RECORD));
-        cast.getArguments().add(new MutableObject<ILogicalExpression>(new VariableReferenceExpression(inputRecordVar)));
-        TypeComputerUtilities.setRequiredAndInputTypes(cast, requiredRecordType, inputRecordType);
-        LogicalVariable newAssignVar = context.newVar();
-        AssignOperator newAssignOperator = new AssignOperator(newAssignVar, new MutableObject<ILogicalExpression>(cast));
-        newAssignOperator.getInputs().add(new MutableObject<ILogicalOperator>(op3));
-
-        List<LogicalVariable> projectVariables = new ArrayList<LogicalVariable>();
-        VariableUtilities.getProducedVariables(oldAssignOperator, projectVariables);
-        projectVariables.add(newAssignVar);
-        ProjectOperator projectOperator = new ProjectOperator(projectVariables);
-        projectOperator.getInputs().add(new MutableObject<ILogicalOperator>(newAssignOperator));
-
-        ILogicalExpression payloadExpr = new VariableReferenceExpression(newAssignVar);
-        MutableObject<ILogicalExpression> payloadRef = new MutableObject<ILogicalExpression>(payloadExpr);
-        InsertDeleteOperator newInserDeleteOperator = new InsertDeleteOperator(insertDeleteOperator.getDataSource(),
-                payloadRef, insertDeleteOperator.getPrimaryKeyExpressions(), insertDeleteOperator.getOperation());
-        newInserDeleteOperator.getInputs().add(new MutableObject<ILogicalOperator>(projectOperator));
-        insertDeleteOperator.getInputs().clear();
-        op1.getInputs().get(0).setValue(newInserDeleteOperator);
-
-        context.computeAndSetTypeEnvironmentForOperator(newAssignOperator);
-        context.computeAndSetTypeEnvironmentForOperator(projectOperator);
-        context.computeAndSetTypeEnvironmentForOperator(newInserDeleteOperator);
-        return true;
+        LogicalVariable replacedVar = addCast(requiredRecordType, recordVar, insertDeleteOp, context);
+        return replacedVar == null;
     }
 
+    /**
+     * Inject a cast-record function when necessary
+     * 
+     * @param requiredRecordType
+     *            the required record type
+     * @param recordVar
+     *            the record variable
+     * @param parent
+     *            the current parent operator to be rewritten
+     * @param context
+     *            the optimization context
+     * @return true if cast is injected; false otherwise.
+     * @throws AlgebricksException
+     */
+    public LogicalVariable addCast(ARecordType requiredRecordType, LogicalVariable recordVar, ILogicalOperator parent,
+            IOptimizationContext context) throws AlgebricksException {
+        List<Mutable<ILogicalOperator>> opRefs = parent.getInputs();
+        for (int index = 0; index < opRefs.size(); index++) {
+            Mutable<ILogicalOperator> opRef = opRefs.get(index);
+            ILogicalOperator op = opRef.getValue();
+
+            List<LogicalVariable> producedVars = new ArrayList<LogicalVariable>();
+            VariableUtilities.getProducedVariables(op, producedVars);
+            IVariableTypeEnvironment env = op.computeOutputTypeEnvironment(context);
+
+            for (int i = 0; i < producedVars.size(); i++) {
+                LogicalVariable var = producedVars.get(i);
+                if (var.equals(recordVar)) {
+                    IAType actualType = (IAType) env.getVarType(var);
+                    if (!capatible(requiredRecordType, actualType)) {
+                        AbstractFunctionCallExpression cast = new ScalarFunctionCallExpression(
+                                FunctionUtils.getFunctionInfo(AsterixBuiltinFunctions.CAST_RECORD));
+                        cast.getArguments().add(
+                                new MutableObject<ILogicalExpression>(new VariableReferenceExpression(var)));
+                        TypeComputerUtilities.setRequiredAndInputTypes(cast, requiredRecordType, actualType);
+                        LogicalVariable newAssignVar = context.newVar();
+                        AssignOperator newAssignOperator = new AssignOperator(newAssignVar,
+                                new MutableObject<ILogicalExpression>(cast));
+                        newAssignOperator.getInputs().add(new MutableObject<ILogicalOperator>(op));
+                        opRef.setValue(newAssignOperator);
+                        context.computeAndSetTypeEnvironmentForOperator(parent);
+                        context.computeAndSetTypeEnvironmentForOperator(newAssignOperator);
+                        newAssignOperator.computeOutputTypeEnvironment(context);
+                        VariableUtilities.substituteVariables(parent, recordVar, newAssignVar, context);
+                        return newAssignVar;
+                    }
+                }
+            }
+            LogicalVariable replacedVar = addCast(requiredRecordType, recordVar, op, context);
+            if (replacedVar != null) {
+                VariableUtilities.substituteVariables(parent, recordVar, replacedVar, context);
+                return replacedVar;
+            }
+        }
+        return null;
+    }
+
+    private boolean capatible(ARecordType reqType, IAType inputType) {
+        if (inputType.getTypeTag() == ATypeTag.ANY) {
+            return false;
+        }
+        IAType[] reqTypes = reqType.getFieldTypes();
+        String[] reqFieldNames = reqType.getFieldNames();
+        IAType[] inputTypes = ((ARecordType) inputType).getFieldTypes();
+        String[] inputFieldNames = ((ARecordType) inputType).getFieldNames();
+
+        if (reqTypes.length != inputTypes.length) {
+            return false;
+        }
+        for (int i = 0; i < reqTypes.length; i++) {
+            if (!reqFieldNames[i].equals(inputFieldNames[i])) {
+                return false;
+            }
+            IAType reqTypeInside = reqTypes[i];
+            if (reqTypes[i].getTypeTag() == ATypeTag.UNION
+                    && NonTaggedFormatUtil.isOptionalField((AUnionType) reqTypes[i])) {
+                reqTypeInside = ((AUnionType) reqTypes[i]).getUnionList().get(
+                        NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST);
+            }
+            IAType inputTypeInside = inputTypes[i];
+            if (inputTypes[i].getTypeTag() == ATypeTag.UNION
+                    && NonTaggedFormatUtil.isOptionalField((AUnionType) inputTypes[i])) {
+                inputTypeInside = ((AUnionType) inputTypes[i]).getUnionList().get(
+                        NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST);
+            }
+            if (inputTypeInside.getTypeTag() != ATypeTag.NULL && !reqTypeInside.equals(inputTypeInside)) {
+                return false;
+            }
+        }
+        return true;
+    }
 }
diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/typecast/StaticTypeCastUtil.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/typecast/StaticTypeCastUtil.java
index 6e3ab37..e72c202 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/typecast/StaticTypeCastUtil.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/typecast/StaticTypeCastUtil.java
@@ -189,8 +189,7 @@
         if (TypeComputerUtilities.getRequiredType(funcExpr) != null)
             return false;
         TypeComputerUtilities.setRequiredAndInputTypes(funcExpr, requiredRecordType, inputRecordType);
-        staticRecordTypeCast(funcExpr, requiredRecordType, inputRecordType, env);
-        return true;
+        return staticRecordTypeCast(funcExpr, requiredRecordType, inputRecordType, env);
     }
 
     /**
@@ -245,7 +244,7 @@
      *            The type environment.
      * @throws AlgebricksException
      */
-    private static void staticRecordTypeCast(AbstractFunctionCallExpression func, ARecordType reqType,
+    private static boolean staticRecordTypeCast(AbstractFunctionCallExpression func, ARecordType reqType,
             ARecordType inputType, IVariableTypeEnvironment env) throws AlgebricksException {
         IAType[] reqFieldTypes = reqType.getFieldTypes();
         String[] reqFieldNames = reqType.getFieldNames();
@@ -341,8 +340,10 @@
                 }
             }
             // the input has extra fields
-            if (!matched && !reqType.isOpen())
-                throw new AlgebricksException("static type mismatch: including an extra closed field " + fieldName);
+            if (!matched && !reqType.isOpen()) {
+                throw new AlgebricksException("static type mismatch: including an extra closed field " + fieldName
+                        + " of type " + fieldType);
+            }
         }
 
         // backward match: match from required to actual
@@ -385,7 +386,12 @@
                 nullFields[i] = true;
             } else {
                 // no matched field in the input for a required closed field
-                throw new AlgebricksException("static type mismatch: miss a required closed field " + reqFieldName);
+                if (inputType.isOpen()) {
+                    //if the input type is open, return false, give that to dynamic type cast to defer the error to the runtime
+                    return false;
+                } else {
+                    throw new AlgebricksException("static type mismatch: miss a required closed field " + reqFieldName);
+                }
             }
         }
 
@@ -472,6 +478,7 @@
                 arguments.add(expRef);
             }
         }
+        return true;
     }
 
     /**
@@ -505,7 +512,7 @@
      *            the input type
      * @return true if the two types are compatiable; false otherwise
      */
-    private static boolean compatible(IAType reqType, IAType inputType) {
+    public static boolean compatible(IAType reqType, IAType inputType) {
         if (reqType.getTypeTag() == ATypeTag.ANY || inputType.getTypeTag() == ATypeTag.ANY) {
             return true;
         }
diff --git a/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453-2/query-issue453-2.1.ddl.aql b/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453-2/query-issue453-2.1.ddl.aql
new file mode 100644
index 0000000..6fcce66
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453-2/query-issue453-2.1.ddl.aql
@@ -0,0 +1,20 @@
+/*
+ * Description  : This test case is to verify the fix for issue453
+ 				: https://code.google.com/p/asterixdb/issues/detail?id=453
+ * Expected Res : SUCCESS
+ * Date         : 18th May 2013
+ */
+
+drop dataverse test if exists;
+create dataverse test;
+use dataverse test;
+
+create type TypeOpen as open {
+    id : int32,
+    int_m : int32,
+    int_o : int32?,
+    string_m : string,
+    string_o : string?
+};
+
+create dataset DataOpen(TypeOpen) primary key id;
diff --git a/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453-2/query-issue453-2.2.update.aql b/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453-2/query-issue453-2.2.update.aql
new file mode 100644
index 0000000..2682e84
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453-2/query-issue453-2.2.update.aql
@@ -0,0 +1,27 @@
+/*
+ * Description  : This test case is to verify the fix for issue453
+ 				: https://code.google.com/p/asterixdb/issues/detail?id=453
+ * Expected Res : SUCCESS
+ * Date         : 18th May 2013
+ */
+
+use dataverse test;
+
+insert into dataset DataOpen(
+      for $arr at $pos in (
+          for $i1 in [1, 2]
+          for $i2 in [1, null]
+          for $s1 in ["a", "b"]
+          for $s2 in ["a", null]
+          return
+            [ $i1, $i2, $s1, $s2]
+        )
+      return
+          {
+            "id" : $pos,
+            "int_m" : $arr[0], 
+            "int_o" : $arr[1],
+            "string_m" : $arr[2],
+            "string_o" : $arr[3]
+          }
+      )
diff --git a/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453-2/query-issue453-2.3.query.aql b/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453-2/query-issue453-2.3.query.aql
new file mode 100644
index 0000000..1381365
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453-2/query-issue453-2.3.query.aql
@@ -0,0 +1,12 @@
+/*
+ * Description  : This test case is to verify the fix for issue453
+ 				: https://code.google.com/p/asterixdb/issues/detail?id=453
+ * Expected Res : SUCCESS
+ * Date         : 18th May 2013
+ */
+
+use dataverse test;
+
+for $d in dataset DataOpen
+order by $d.id
+return $d
\ No newline at end of file
diff --git a/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453/query-issue453.1.ddl.aql b/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453/query-issue453.1.ddl.aql
new file mode 100644
index 0000000..6fcce66
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453/query-issue453.1.ddl.aql
@@ -0,0 +1,20 @@
+/*
+ * Description  : This test case is to verify the fix for issue453
+ 				: https://code.google.com/p/asterixdb/issues/detail?id=453
+ * Expected Res : SUCCESS
+ * Date         : 18th May 2013
+ */
+
+drop dataverse test if exists;
+create dataverse test;
+use dataverse test;
+
+create type TypeOpen as open {
+    id : int32,
+    int_m : int32,
+    int_o : int32?,
+    string_m : string,
+    string_o : string?
+};
+
+create dataset DataOpen(TypeOpen) primary key id;
diff --git a/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453/query-issue453.2.update.aql b/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453/query-issue453.2.update.aql
new file mode 100644
index 0000000..3346f52
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453/query-issue453.2.update.aql
@@ -0,0 +1,16 @@
+/*
+ * Description  : This test case is to verify the fix for issue453
+ 				: https://code.google.com/p/asterixdb/issues/detail?id=453
+ * Expected Res : SUCCESS
+ * Date         : 18th May 2013
+ */
+
+use dataverse test;
+
+insert into dataset DataOpen(
+      for $o in {{
+          { "id": 0, "int_m": 1, "int_o": 1, "string_m": "a", "string_o": "a" },
+          { "id": 1, "int_m": 1, "int_o": 1, "string_m": "a", "string_o": null }
+        }}
+      return $o
+)
\ No newline at end of file
diff --git a/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453/query-issue453.3.query.aql b/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453/query-issue453.3.query.aql
new file mode 100644
index 0000000..1381365
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/open-closed/query-issue453/query-issue453.3.query.aql
@@ -0,0 +1,12 @@
+/*
+ * Description  : This test case is to verify the fix for issue453
+ 				: https://code.google.com/p/asterixdb/issues/detail?id=453
+ * Expected Res : SUCCESS
+ * Date         : 18th May 2013
+ */
+
+use dataverse test;
+
+for $d in dataset DataOpen
+order by $d.id
+return $d
\ No newline at end of file
diff --git a/asterix-app/src/test/resources/runtimets/results/open-closed/query-issue453-2/query-issue453-2.1.adm b/asterix-app/src/test/resources/runtimets/results/open-closed/query-issue453-2/query-issue453-2.1.adm
new file mode 100644
index 0000000..b573845
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/results/open-closed/query-issue453-2/query-issue453-2.1.adm
@@ -0,0 +1,16 @@
+{ "id": 0, "int_m": 1, "int_o": 1, "string_m": "a", "string_o": "a" }
+{ "id": 1, "int_m": 1, "int_o": 1, "string_m": "a", "string_o": null }
+{ "id": 2, "int_m": 1, "int_o": 1, "string_m": "b", "string_o": "a" }
+{ "id": 3, "int_m": 1, "int_o": 1, "string_m": "b", "string_o": null }
+{ "id": 4, "int_m": 1, "int_o": null, "string_m": "a", "string_o": "a" }
+{ "id": 5, "int_m": 1, "int_o": null, "string_m": "a", "string_o": null }
+{ "id": 6, "int_m": 1, "int_o": null, "string_m": "b", "string_o": "a" }
+{ "id": 7, "int_m": 1, "int_o": null, "string_m": "b", "string_o": null }
+{ "id": 8, "int_m": 2, "int_o": 1, "string_m": "a", "string_o": "a" }
+{ "id": 9, "int_m": 2, "int_o": 1, "string_m": "a", "string_o": null }
+{ "id": 10, "int_m": 2, "int_o": 1, "string_m": "b", "string_o": "a" }
+{ "id": 11, "int_m": 2, "int_o": 1, "string_m": "b", "string_o": null }
+{ "id": 12, "int_m": 2, "int_o": null, "string_m": "a", "string_o": "a" }
+{ "id": 13, "int_m": 2, "int_o": null, "string_m": "a", "string_o": null }
+{ "id": 14, "int_m": 2, "int_o": null, "string_m": "b", "string_o": "a" }
+{ "id": 15, "int_m": 2, "int_o": null, "string_m": "b", "string_o": null }
diff --git a/asterix-app/src/test/resources/runtimets/results/open-closed/query-issue453/query-issue453.1.adm b/asterix-app/src/test/resources/runtimets/results/open-closed/query-issue453/query-issue453.1.adm
new file mode 100644
index 0000000..c076685
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/results/open-closed/query-issue453/query-issue453.1.adm
@@ -0,0 +1,2 @@
+{ "id": 0, "int_m": 1, "int_o": 1, "string_m": "a", "string_o": "a" }
+{ "id": 1, "int_m": 1, "int_o": 1, "string_m": "a", "string_o": null }
diff --git a/asterix-app/src/test/resources/runtimets/testsuite.xml b/asterix-app/src/test/resources/runtimets/testsuite.xml
index 976f071..ff3126a 100644
--- a/asterix-app/src/test/resources/runtimets/testsuite.xml
+++ b/asterix-app/src/test/resources/runtimets/testsuite.xml
@@ -2730,6 +2730,16 @@
         <expected-error>edu.uci.ics.asterix.common.exceptions.AsterixException</expected-error>
       </compilation-unit>
     </test-case>
+    <test-case FilePath="open-closed">
+      <compilation-unit name="query-issue453">
+	<output-dir compare="Text">query-issue453</output-dir>
+      </compilation-unit>
+    </test-case>
+    <test-case FilePath="open-closed">
+      <compilation-unit name="query-issue453-2">
+	<output-dir compare="Text">query-issue453-2</output-dir>
+      </compilation-unit>
+    </test-case>
   </test-group>
   <test-group name="quantifiers">
     <test-case FilePath="quantifiers">
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/RecordBuilder.java b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/RecordBuilder.java
index f5d07ae..10b6071 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/RecordBuilder.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/RecordBuilder.java
@@ -150,7 +150,7 @@
         // +1 because we do not store the value tag.
         closedPartOutputStream.write(value.getByteArray(), value.getStartOffset() + 1, len);
         numberOfClosedFields++;
-        if (isNullable && value.getByteArray()[0] != SER_NULL_TYPE_TAG) {
+        if (isNullable && value.getByteArray()[value.getStartOffset()] != SER_NULL_TYPE_TAG) {
             nullBitMap[id / 8] |= (byte) (1 << (7 - (id % 8)));
         }
     }
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 494ea6f..ed9690c 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
@@ -99,7 +99,7 @@
             start = bos.size();
             dos.write(ATypeTag.NULL.serialize());
             end = bos.size();
-            nullTypeTag.set(bos.getByteArray(), start, end);
+            nullTypeTag.set(bos.getByteArray(), start, end - start);
         } catch (IOException e) {
             throw new IllegalStateException(e);
         }
@@ -150,7 +150,7 @@
         for (int i = 0; i < optionalFields.length; i++)
             optionalFields[i] = false;
 
-        bos.reset(nullReference.getStartOffset() + nullReference.getLength());
+        bos.reset(nullTypeTag.getStartOffset() + nullTypeTag.getLength());
         for (int i = 0; i < numSchemaFields; i++) {
             ATypeTag ftypeTag = fieldTypes[i].getTypeTag();
             String fname = fieldNames[i];
@@ -278,8 +278,14 @@
             // recursively casting, the result of casting can always be thought
             // as flat
             if (optionalFields[i]) {
-                nestedVisitorArg.second = ((AUnionType) fType).getUnionList().get(
-                        NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST);
+                if (fieldTypeTags.size() <= i || fieldTypeTags.get(i) == null
+                        || fieldTypeTags.get(i).equals(nullTypeTag)) {
+                    //the field is optional in the input record
+                    nestedVisitorArg.second = ((AUnionType) fType).getUnionList().get(0);
+                } else {
+                    nestedVisitorArg.second = ((AUnionType) fType).getUnionList().get(
+                            NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST);
+                }
             }
             field.accept(visitor, nestedVisitorArg);
             recBuilder.addField(i, nestedVisitorArg.first);