diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/base/RuleCollections.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/base/RuleCollections.java
index aa22be5..43fdf4b 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/base/RuleCollections.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/base/RuleCollections.java
@@ -15,6 +15,7 @@
 import edu.uci.ics.asterix.optimizer.rules.IfElseToSwitchCaseFunctionRule;
 import edu.uci.ics.asterix.optimizer.rules.InlineAssignIntoAggregateRule;
 import edu.uci.ics.asterix.optimizer.rules.IntroduceBTreeIndexSearchRule;
+import edu.uci.ics.asterix.optimizer.rules.IntroduceCastRecordRule;
 import edu.uci.ics.asterix.optimizer.rules.IntroduceRTreeIndexSearchRule;
 import edu.uci.ics.asterix.optimizer.rules.IntroduceSecondaryIndexInsertDeleteRule;
 import edu.uci.ics.asterix.optimizer.rules.LoadRecordFieldsRule;
@@ -81,7 +82,11 @@
         normalization.add(new ExtractGbyExpressionsRule());
         normalization.add(new ExtractDistinctByExpressionsRule());
         normalization.add(new ExtractOrderExpressionsRule());
+
+        // TopdownTypeInferenceRule should go before IntroduceCastRecordRule to
+        // avoid unnecessary casting
         normalization.add(new TopDownTypeInferenceRule());
+        normalization.add(new IntroduceCastRecordRule());
         normalization.add(new ConstantFoldingRule());
         normalization.add(new UnnestToDataScanRule());
         normalization.add(new IfElseToSwitchCaseFunctionRule());
diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/ByNameToByIndexFieldAccessRule.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/ByNameToByIndexFieldAccessRule.java
index e97961f..25412d8 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/ByNameToByIndexFieldAccessRule.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/ByNameToByIndexFieldAccessRule.java
@@ -1,6 +1,7 @@
 package edu.uci.ics.asterix.optimizer.rules;
 
 import java.util.ArrayList;
+import java.util.List;
 
 import org.apache.commons.lang3.mutable.Mutable;
 import org.apache.commons.lang3.mutable.MutableObject;
@@ -41,7 +42,8 @@
     }
 
     @Override
-    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+            throws AlgebricksException {
         AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
         if (op.getOperatorTag() != LogicalOperatorTag.ASSIGN) {
             return false;
@@ -55,64 +57,70 @@
         if (assign.getExpressions().get(0).getValue().getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL) {
             return false;
         }
-        AbstractFunctionCallExpression fce = (AbstractFunctionCallExpression) assign.getExpressions().get(0)
-                .getValue();
-        if (fce.getFunctionIdentifier() != AsterixBuiltinFunctions.FIELD_ACCESS_BY_NAME) {
-            return false;
-        }
-        IVariableTypeEnvironment env = context.getOutputTypeEnvironment(op);
 
-        ILogicalExpression a0 = fce.getArguments().get(0).getValue();
-        if (a0.getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL) {
-            LogicalVariable var1 = context.newVar();
-            ArrayList<LogicalVariable> varArray = new ArrayList<LogicalVariable>(1);
-            varArray.add(var1);
-            ArrayList<Mutable<ILogicalExpression>> exprArray = new ArrayList<Mutable<ILogicalExpression>>(1);
-            exprArray.add(new MutableObject<ILogicalExpression>(a0));
-            AssignOperator assignVar = new AssignOperator(varArray, exprArray);
-            fce.getArguments().get(0).setValue(new VariableReferenceExpression(var1));
-            assignVar.getInputs().add(new MutableObject<ILogicalOperator>(assign.getInputs().get(0).getValue()));
-            assign.getInputs().get(0).setValue(assignVar);
-            context.computeAndSetTypeEnvironmentForOperator(assignVar);
-            context.computeAndSetTypeEnvironmentForOperator(assign);
-        }
-
-        IAType t = (IAType) env.getType(fce.getArguments().get(0).getValue());
-        switch (t.getTypeTag()) {
-            case ANY: {
-                return false;
+        List<Mutable<ILogicalExpression>> expressions = assign.getExpressions();
+        boolean changed = false;
+        for (int i = 0; i < expressions.size(); i++) {
+            AbstractFunctionCallExpression fce = (AbstractFunctionCallExpression) expressions.get(i).getValue();
+            if (fce.getFunctionIdentifier() != AsterixBuiltinFunctions.FIELD_ACCESS_BY_NAME) {
+                continue;
             }
-            case RECORD: {
-                ARecordType recType = (ARecordType) t;
-                ILogicalExpression fai = createFieldAccessByIndex(recType, fce);
-                if (fai == null) {
+            IVariableTypeEnvironment env = context.getOutputTypeEnvironment(op);
+
+            ILogicalExpression a0 = fce.getArguments().get(0).getValue();
+            if (a0.getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL) {
+                LogicalVariable var1 = context.newVar();
+                ArrayList<LogicalVariable> varArray = new ArrayList<LogicalVariable>(1);
+                varArray.add(var1);
+                ArrayList<Mutable<ILogicalExpression>> exprArray = new ArrayList<Mutable<ILogicalExpression>>(1);
+                exprArray.add(new MutableObject<ILogicalExpression>(a0));
+                AssignOperator assignVar = new AssignOperator(varArray, exprArray);
+                fce.getArguments().get(0).setValue(new VariableReferenceExpression(var1));
+                assignVar.getInputs().add(new MutableObject<ILogicalOperator>(assign.getInputs().get(0).getValue()));
+                assign.getInputs().get(0).setValue(assignVar);
+                context.computeAndSetTypeEnvironmentForOperator(assignVar);
+                context.computeAndSetTypeEnvironmentForOperator(assign);
+            }
+
+            IAType t = (IAType) env.getType(fce.getArguments().get(0).getValue());
+            switch (t.getTypeTag()) {
+                case ANY: {
                     return false;
                 }
-                assign.getExpressions().get(0).setValue(fai);
-                break;
-            }
-            case UNION: {
-                AUnionType unionT = (AUnionType) t;
-                if (unionT.isNullableType()) {
-                    IAType t2 = unionT.getUnionList().get(1);
-                    if (t2.getTypeTag() == ATypeTag.RECORD) {
-                        ARecordType recType = (ARecordType) t2;
-                        ILogicalExpression fai = createFieldAccessByIndex(recType, fce);
-                        if (fai == null) {
-                            return false;
-                        }
-                        assign.getExpressions().get(0).setValue(fai);
-                        break;
+                case RECORD: {
+                    ARecordType recType = (ARecordType) t;
+                    ILogicalExpression fai = createFieldAccessByIndex(recType, fce);
+                    if (fai == null) {
+                        return false;
                     }
+                    expressions.get(i).setValue(fai);
+                    changed = true;
+                    break;
                 }
-                throw new NotImplementedException("Union " + unionT);
-            }
-            default: {
-                throw new AlgebricksException("Cannot call field-access on data of type " + t);
+                case UNION: {
+                    AUnionType unionT = (AUnionType) t;
+                    if (unionT.isNullableType()) {
+                        IAType t2 = unionT.getUnionList().get(1);
+                        if (t2.getTypeTag() == ATypeTag.RECORD) {
+                            ARecordType recType = (ARecordType) t2;
+                            ILogicalExpression fai = createFieldAccessByIndex(recType, fce);
+                            if (fai == null) {
+                                return false;
+                            }
+                            expressions.get(i).setValue(fai);
+                            changed = true;
+                            break;
+                        }
+                    }
+                    throw new NotImplementedException("Union " + unionT);
+                }
+                default: {
+                    throw new AlgebricksException("Cannot call field-access on data of type " + t);
+                }
             }
         }
         assign.removeAnnotation(AsterixOperatorAnnotations.PUSHED_FIELD_ACCESS);
-        return true;
+        return changed;
     }
 
     private static ILogicalExpression createFieldAccessByIndex(ARecordType recType, AbstractFunctionCallExpression fce) {
diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/ConstantFoldingRule.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/ConstantFoldingRule.java
index faae5eb..36783ce 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/ConstantFoldingRule.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/ConstantFoldingRule.java
@@ -147,7 +147,8 @@
                 return new Pair<Boolean, ILogicalExpression>(changed, expr);
             }
             // TODO: currently ARecord is always a closed record
-            if (expr.getFunctionIdentifier().equals(AsterixBuiltinFunctions.OPEN_RECORD_CONSTRUCTOR)) {
+            if (expr.getFunctionIdentifier().equals(AsterixBuiltinFunctions.OPEN_RECORD_CONSTRUCTOR)
+                    || expr.getFunctionIdentifier().equals(AsterixBuiltinFunctions.CAST_RECORD)) {
                 return new Pair<Boolean, ILogicalExpression>(false, null);
             }
             if (expr.getFunctionIdentifier().equals(AsterixBuiltinFunctions.FIELD_ACCESS_BY_NAME)) {
diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceCastRecordRule.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceCastRecordRule.java
new file mode 100644
index 0000000..fc96940
--- /dev/null
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceCastRecordRule.java
@@ -0,0 +1,122 @@
+package edu.uci.ics.asterix.optimizer.rules;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.commons.lang3.mutable.MutableObject;
+
+import edu.uci.ics.asterix.aql.util.FunctionUtils;
+import edu.uci.ics.asterix.metadata.declared.AqlDataSource;
+import edu.uci.ics.asterix.om.functions.AsterixBuiltinFunctions;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.IAType;
+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;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
+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;
+
+public class IntroduceCastRecordRule implements IAlgebraicRewriteRule {
+
+    @Override
+    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {
+        return false;
+    }
+
+    @Override
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+            throws AlgebricksException {
+        /**
+         * pattern match: sink insert assign
+         * 
+         * resulting plan: sink-insert-project-assign
+         * 
+         */
+        AbstractLogicalOperator op1 = (AbstractLogicalOperator) opRef.getValue();
+        if (op1.getOperatorTag() != LogicalOperatorTag.SINK)
+            return false;
+        AbstractLogicalOperator op2 = (AbstractLogicalOperator) op1.getInputs().get(0).getValue();
+        if (op2.getOperatorTag() != LogicalOperatorTag.INSERT_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);
+        ARecordType inputRecordType = (ARecordType) env.getVarType(inputRecordVar);
+
+        boolean needCast = needCast(requiredRecordType, inputRecordType);
+        if (needCast) {
+            // insert
+            // project
+            // assign
+            // assign
+            AbstractFunctionCallExpression cast = new ScalarFunctionCallExpression(
+                    FunctionUtils.getFunctionInfo(AsterixBuiltinFunctions.CAST_RECORD));
+            ARecordType[] types = new ARecordType[2];
+            types[0] = requiredRecordType;
+            types[1] = inputRecordType;
+            cast.getArguments().add(
+                    new MutableObject<ILogicalExpression>(new VariableReferenceExpression(inputRecordVar)));
+            cast.setOpaqueParameters(types);
+            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);
+            return true;
+        }
+        return false;
+
+    }
+
+    private boolean needCast(ARecordType reqType, ARecordType inputType) {
+        if (!reqType.equals(inputType))
+            return true;
+        else
+            return false;
+    }
+
+}
diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/LoadRecordFieldsRule.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/LoadRecordFieldsRule.java
index d5deb3c..0a68315 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/LoadRecordFieldsRule.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/LoadRecordFieldsRule.java
@@ -50,11 +50,13 @@
     }
 
     @Override
-    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+            throws AlgebricksException {
         AbstractLogicalOperator op1 = (AbstractLogicalOperator) opRef.getValue();
         if (context.checkIfInDontApplySet(this, op1)) {
             return false;
         }
+
         if (op1.getOperatorTag() == LogicalOperatorTag.ASSIGN) {
             AssignOperator a1 = (AssignOperator) op1;
             ILogicalExpression expr = getFirstExpr(a1);
@@ -209,7 +211,7 @@
      * @param toPushThroughChildRef
      */
     private static void pushAccessAboveOpRef(AssignOperator toPush, Mutable<ILogicalOperator> toPushThroughChildRef,
-            IOptimizationContext context) {
+            IOptimizationContext context) throws AlgebricksException {
         List<Mutable<ILogicalOperator>> tpInpList = toPush.getInputs();
         tpInpList.clear();
         tpInpList.add(new MutableObject<ILogicalOperator>(toPushThroughChildRef.getValue()));
@@ -233,7 +235,7 @@
      * 
      * @param toPush
      */
-    private static boolean findAndEliminateRedundantFieldAccess(AssignOperator assign) {
+    private static boolean findAndEliminateRedundantFieldAccess(AssignOperator assign) throws AlgebricksException {
         ILogicalExpression expr = getFirstExpr(assign);
         AbstractFunctionCallExpression f = (AbstractFunctionCallExpression) expr;
         ILogicalExpression arg0 = f.getArguments().get(0).getValue();
@@ -250,9 +252,20 @@
         if (f.getFunctionIdentifier().equals(AsterixBuiltinFunctions.FIELD_ACCESS_BY_NAME)) {
             String fldName = ((AString) ((AsterixConstantValue) ce.getValue()).getObject()).getStringValue();
             ILogicalExpression fldExpr = findFieldExpression(assign, recordVar, fldName);
+
             if (fldExpr != null) {
-                assign.getExpressions().get(0).setValue(fldExpr);
-                return true;
+                // check the liveness of the new expression
+                List<LogicalVariable> usedVariables = new ArrayList<LogicalVariable>();
+                fldExpr.getUsedVariables(usedVariables);
+                List<LogicalVariable> liveInputVars = new ArrayList<LogicalVariable>();
+                VariableUtilities.getLiveVariables(assign, liveInputVars);
+                usedVariables.removeAll(liveInputVars);
+                if (usedVariables.size() == 0) {
+                    assign.getExpressions().get(0).setValue(fldExpr);
+                    return true;
+                } else {
+                    return false;
+                }
             } else {
                 return false;
             }
diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/TopDownTypeInferenceRule.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/TopDownTypeInferenceRule.java
index 958f5f6..08b50cc 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/TopDownTypeInferenceRule.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/TopDownTypeInferenceRule.java
@@ -4,11 +4,19 @@
 import java.util.List;
 
 import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.commons.lang3.mutable.MutableObject;
 
 import edu.uci.ics.asterix.metadata.declared.AqlDataSource;
+import edu.uci.ics.asterix.om.base.ANull;
+import edu.uci.ics.asterix.om.base.AString;
+import edu.uci.ics.asterix.om.constants.AsterixConstantValue;
 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.BuiltinType;
 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;
@@ -17,6 +25,7 @@
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
 import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
 import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
 import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
 import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
@@ -36,9 +45,12 @@
     public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
             throws AlgebricksException {
         /**
-         * pattern match: sink/insert/assign
-         * record type is propagated from insert data source to the record-constructor expression
+         * pattern match: sink/insert/assign record type is propagated from
+         * insert data source to the record-constructor expression
          */
+        if (context.checkIfInDontApplySet(this, opRef.getValue()))
+            return false;
+        context.addToDontApplySet(this, opRef.getValue());
         AbstractLogicalOperator op1 = (AbstractLogicalOperator) opRef.getValue();
         if (op1.getOperatorTag() != LogicalOperatorTag.SINK)
             return false;
@@ -73,27 +85,31 @@
         AbstractLogicalOperator currentOperator = oldAssignOperator;
         List<LogicalVariable> producedVariables = new ArrayList<LogicalVariable>();
         boolean changed = false;
-        if (!requiredRecordType.equals(inputRecordType)) {
-            /**
-             * find the assign operator for the "input record" to the
-             * insert_delete operator
-             */
-            do {
-                if (currentOperator.getOperatorTag() == LogicalOperatorTag.ASSIGN) {
-                    producedVariables.clear();
-                    VariableUtilities.getProducedVariables(currentOperator, producedVariables);
-                    int position = producedVariables.indexOf(oldRecordVariable);
 
-                    /**
-                     * set the top-down propagated type
-                     */
-                    if (position >= 0) {
-                        AssignOperator originalAssign = (AssignOperator) currentOperator;
-                        List<Mutable<ILogicalExpression>> expressionPointers = originalAssign.getExpressions();
-                        ILogicalExpression expr = expressionPointers.get(position).getValue();
-                        if (expr.getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL) {
-                            ScalarFunctionCallExpression funcExpr = (ScalarFunctionCallExpression) expr;
-                            changed = TypeComputerUtilities.setRequiredType(funcExpr, requiredRecordType);
+        /**
+         * find the assign operator for the "input record" to the insert_delete
+         * operator
+         */
+        do {
+            if (currentOperator.getOperatorTag() == LogicalOperatorTag.ASSIGN) {
+                producedVariables.clear();
+                VariableUtilities.getProducedVariables(currentOperator, producedVariables);
+                int position = producedVariables.indexOf(oldRecordVariable);
+
+                /**
+                 * set the top-down propagated type
+                 */
+                if (position >= 0) {
+                    AssignOperator originalAssign = (AssignOperator) currentOperator;
+                    List<Mutable<ILogicalExpression>> expressionPointers = originalAssign.getExpressions();
+                    ILogicalExpression expr = expressionPointers.get(position).getValue();
+                    if (expr.getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL) {
+                        ScalarFunctionCallExpression funcExpr = (ScalarFunctionCallExpression) expr;
+                        changed = TypeComputerUtilities.setRequiredAndInputTypes(funcExpr, requiredRecordType,
+                                inputRecordType);
+                        changed &= !requiredRecordType.equals(inputRecordType);
+                        if (changed) {
+                            staticTypeCast(funcExpr, requiredRecordType, inputRecordType);
                             List<Mutable<ILogicalExpression>> args = funcExpr.getArguments();
                             int openPartStart = requiredRecordType.getFieldTypes().length * 2;
                             for (int j = openPartStart; j < args.size(); j++) {
@@ -105,13 +121,137 @@
                             }
                         }
                     }
+                    context.computeAndSetTypeEnvironmentForOperator(originalAssign);
                 }
-                if (currentOperator.getInputs().size() > 0)
-                    currentOperator = (AbstractLogicalOperator) currentOperator.getInputs().get(0).getValue();
-                else
-                    break;
-            } while (currentOperator != null);
-        }
+            }
+            if (currentOperator.getInputs().size() > 0)
+                currentOperator = (AbstractLogicalOperator) currentOperator.getInputs().get(0).getValue();
+            else
+                break;
+        } while (currentOperator != null);
         return changed;
     }
+
+    private void staticTypeCast(ScalarFunctionCallExpression func, ARecordType reqType, ARecordType inputType) {
+        IAType[] reqFieldTypes = reqType.getFieldTypes();
+        String[] reqFieldNames = reqType.getFieldNames();
+        IAType[] inputFieldTypes = inputType.getFieldTypes();
+        String[] inputFieldNames = inputType.getFieldNames();
+
+        int[] fieldPermutation = new int[reqFieldTypes.length];
+        boolean[] nullFields = new boolean[reqFieldTypes.length];
+        boolean[] openFields = new boolean[inputFieldTypes.length];
+
+        for (int i = 0; i < nullFields.length; i++)
+            nullFields[i] = false;
+        for (int i = 0; i < openFields.length; i++)
+            openFields[i] = true;
+        for (int i = 0; i < fieldPermutation.length; i++)
+            fieldPermutation[i] = -1;
+
+        // forward match: match from actual to required
+        boolean matched = false;
+        for (int i = 0; i < inputFieldNames.length; i++) {
+            String fieldName = inputFieldNames[i];
+            IAType fieldType = inputFieldTypes[i];
+            matched = false;
+            for (int j = 0; j < reqFieldNames.length; j++) {
+                String reqFieldName = reqFieldNames[j];
+                IAType reqFieldType = reqFieldTypes[j];
+                if (fieldName.equals(reqFieldName)) {
+                    if (fieldType.equals(reqFieldType)) {
+                        fieldPermutation[j] = i;
+                        openFields[i] = false;
+                        matched = true;
+                        break;
+                    }
+
+                    // match the optional field
+                    if (reqFieldType.getTypeTag() == ATypeTag.UNION
+                            && NonTaggedFormatUtil.isOptionalField((AUnionType) reqFieldType)) {
+                        IAType itemType = ((AUnionType) reqFieldType).getUnionList().get(
+                                NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST);
+                        if (fieldType.equals(BuiltinType.ANULL) || fieldType.equals(itemType)) {
+                            fieldPermutation[j] = i;
+                            openFields[i] = false;
+                            matched = true;
+                            break;
+                        }
+                    }
+                }
+            }
+            if (matched)
+                continue;
+            // the input has extra fields
+            if (!reqType.isOpen())
+                throw new IllegalStateException("static type mismatch: including extra closed fields");
+        }
+
+        // backward match: match from required to actual
+        for (int i = 0; i < reqFieldNames.length; i++) {
+            String reqFieldName = reqFieldNames[i];
+            IAType reqFieldType = reqFieldTypes[i];
+            matched = false;
+            for (int j = 0; j < inputFieldNames.length; j++) {
+                String fieldName = inputFieldNames[j];
+                IAType fieldType = inputFieldTypes[j];
+                if (fieldName.equals(reqFieldName)) {
+                    if (fieldType.equals(reqFieldType)) {
+                        matched = true;
+                        break;
+                    }
+
+                    // match the optional field
+                    if (reqFieldType.getTypeTag() == ATypeTag.UNION
+                            && NonTaggedFormatUtil.isOptionalField((AUnionType) reqFieldType)) {
+                        IAType itemType = ((AUnionType) reqFieldType).getUnionList().get(
+                                NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST);
+                        if (fieldType.equals(BuiltinType.ANULL) || fieldType.equals(itemType)) {
+                            matched = true;
+                            break;
+                        }
+                    }
+                }
+            }
+            if (matched)
+                continue;
+
+            IAType t = reqFieldTypes[i];
+            if (t.getTypeTag() == ATypeTag.UNION && NonTaggedFormatUtil.isOptionalField((AUnionType) t)) {
+                // add a null field
+                nullFields[i] = true;
+            } else {
+                // no matched field in the input for a required closed field
+                throw new IllegalStateException("static type mismatch: miss a required closed field");
+            }
+        }
+
+        List<Mutable<ILogicalExpression>> arguments = func.getArguments();
+        List<Mutable<ILogicalExpression>> argumentsClone = new ArrayList<Mutable<ILogicalExpression>>();
+        argumentsClone.addAll(arguments);
+        arguments.clear();
+        // re-order the closed part and fill in null fields
+        for (int i = 0; i < fieldPermutation.length; i++) {
+            int pos = fieldPermutation[i];
+            if (pos >= 0) {
+                arguments.add(argumentsClone.get(2 * pos));
+                arguments.add(argumentsClone.get(2 * pos + 1));
+            }
+            if (nullFields[i]) {
+                // add a null field
+                arguments.add(new MutableObject<ILogicalExpression>(new ConstantExpression(new AsterixConstantValue(
+                        new AString(reqFieldNames[i])))));
+                arguments.add(new MutableObject<ILogicalExpression>(new ConstantExpression(new AsterixConstantValue(
+                        ANull.NULL))));
+            }
+        }
+
+        // add the open part
+        for (int i = 0; i < openFields.length; i++) {
+            if (openFields[i]) {
+                arguments.add(argumentsClone.get(2 * i));
+                arguments.add(argumentsClone.get(2 * i + 1));
+            }
+        }
+    }
 }
diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/translator/AqlExpressionToPlanTranslator.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/translator/AqlExpressionToPlanTranslator.java
index 4fd6dde..b1970ac 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/translator/AqlExpressionToPlanTranslator.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/translator/AqlExpressionToPlanTranslator.java
@@ -81,6 +81,7 @@
 import edu.uci.ics.asterix.metadata.MetadataTransactionContext;
 import edu.uci.ics.asterix.metadata.bootstrap.AsterixProperties;
 import edu.uci.ics.asterix.metadata.declared.AqlCompiledDatasetDecl;
+import edu.uci.ics.asterix.metadata.declared.AqlCompiledInternalDatasetDetails;
 import edu.uci.ics.asterix.metadata.declared.AqlCompiledMetadataDeclarations;
 import edu.uci.ics.asterix.metadata.declared.AqlDataSource;
 import edu.uci.ics.asterix.metadata.declared.AqlLogicalPlanAndMetadataImpl;
@@ -88,7 +89,6 @@
 import edu.uci.ics.asterix.metadata.declared.AqlSourceId;
 import edu.uci.ics.asterix.metadata.declared.FileSplitDataSink;
 import edu.uci.ics.asterix.metadata.declared.FileSplitSinkId;
-import edu.uci.ics.asterix.metadata.utils.DatasetUtils;
 import edu.uci.ics.asterix.om.base.AInt32;
 import edu.uci.ics.asterix.om.base.AString;
 import edu.uci.ics.asterix.om.constants.AsterixConstantValue;
@@ -142,7 +142,6 @@
 import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.WriteOperator;
 import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.WriteResultOperator;
 import edu.uci.ics.hyracks.algebricks.core.algebra.plan.ALogicalPlanImpl;
-import edu.uci.ics.hyracks.algebricks.runtime.base.IEvaluatorFactory;
 import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
 
@@ -150,10 +149,9 @@
  * Each visit returns a pair of an operator and a variable. The variable
  * corresponds to the new column, if any, added to the tuple flow. E.g., for
  * Unnest, the column is the variable bound to the elements in the list, for
- * Subplan it is null.
- * The first argument of a visit method is the expression which is translated.
- * The second argument of a visit method is the tuple source for the current
- * subtree.
+ * Subplan it is null. The first argument of a visit method is the expression
+ * which is translated. The second argument of a visit method is the tuple
+ * source for the current subtree.
  */
 
 public class AqlExpressionToPlanTranslator extends AbstractAqlTranslator implements
@@ -236,9 +234,18 @@
             ArrayList<LogicalVariable> vars = new ArrayList<LogicalVariable>();
             ArrayList<Mutable<ILogicalExpression>> exprs = new ArrayList<Mutable<ILogicalExpression>>();
             List<Mutable<ILogicalExpression>> varRefsForLoading = new ArrayList<Mutable<ILogicalExpression>>();
-            for (Triple<IEvaluatorFactory, ScalarFunctionCallExpression, IAType> partitioner : DatasetUtils
-                    .getPartitioningFunctions(adecl)) {
-                AbstractFunctionCallExpression f = partitioner.second.cloneExpression();
+
+            AqlCompiledInternalDatasetDetails datasetDetails = (AqlCompiledInternalDatasetDetails) adecl
+                    .getAqlCompiledDatasetDetails();
+            List<String> partitionKeys = datasetDetails.getPartitioningExprs();
+            for (String keyFieldName : partitionKeys) {
+                IFunctionInfo finfoAccess = AsterixBuiltinFunctions
+                        .getAsterixFunctionInfo(AsterixBuiltinFunctions.FIELD_ACCESS_BY_NAME);
+                @SuppressWarnings("unchecked")
+                ScalarFunctionCallExpression f = new ScalarFunctionCallExpression(finfoAccess,
+                        new MutableObject<ILogicalExpression>(new VariableReferenceExpression(METADATA_DUMMY_VAR)),
+                        new MutableObject<ILogicalExpression>(new ConstantExpression(new AsterixConstantValue(new AString(
+                                keyFieldName)))));
                 f.substituteVar(METADATA_DUMMY_VAR, resVar);
                 exprs.add(new MutableObject<ILogicalExpression>(f));
                 LogicalVariable v = context.newVar();
@@ -401,8 +408,8 @@
             Mutable<ILogicalOperator> tupSource) throws AsterixException {
         Pair<ILogicalExpression, Mutable<ILogicalOperator>> p = aqlExprToAlgExpression(fa.getExpr(), tupSource);
         LogicalVariable v = context.newVar();
-        AbstractFunctionCallExpression fldAccess = new ScalarFunctionCallExpression(FunctionUtils
-                .getFunctionInfo(AsterixBuiltinFunctions.FIELD_ACCESS_BY_NAME));
+        AbstractFunctionCallExpression fldAccess = new ScalarFunctionCallExpression(
+                FunctionUtils.getFunctionInfo(AsterixBuiltinFunctions.FIELD_ACCESS_BY_NAME));
         fldAccess.getArguments().add(new MutableObject<ILogicalExpression>(p.first));
         ILogicalExpression faExpr = new ConstantExpression(new AsterixConstantValue(new AString(fa.getIdent()
                 .getValue())));
@@ -421,8 +428,8 @@
         AbstractFunctionCallExpression f;
         int i = ia.getIndex();
         if (i == IndexAccessor.ANY) {
-            f = new ScalarFunctionCallExpression(FunctionUtils
-                    .getFunctionInfo(AsterixBuiltinFunctions.ANY_COLLECTION_MEMBER));
+            f = new ScalarFunctionCallExpression(
+                    FunctionUtils.getFunctionInfo(AsterixBuiltinFunctions.ANY_COLLECTION_MEMBER));
             f.getArguments().add(new MutableObject<ILogicalExpression>(p.first));
         } else {
             f = new ScalarFunctionCallExpression(FunctionUtils.getFunctionInfo(AsterixBuiltinFunctions.GET_ITEM));
@@ -469,7 +476,8 @@
             }
         }
 
-        FunctionIdentifier fi = new FunctionIdentifier(AlgebricksBuiltinFunctions.ALGEBRICKS_NS, fid.getFunctionName(), false);
+        FunctionIdentifier fi = new FunctionIdentifier(AlgebricksBuiltinFunctions.ALGEBRICKS_NS, fid.getFunctionName(),
+                false);
         AsterixFunctionInfo afi = AsterixBuiltinFunctions.lookupFunction(fi);
         FunctionIdentifier builtinAquafi = afi == null ? null : afi.getFunctionIdentifier();
 
@@ -486,8 +494,8 @@
         if (AsterixBuiltinFunctions.isBuiltinAggregateFunction(fi)) {
             f = AsterixBuiltinFunctions.makeAggregateFunctionExpression(fi, args);
         } else if (AsterixBuiltinFunctions.isBuiltinUnnestingFunction(fi)) {
-            UnnestingFunctionCallExpression ufce = new UnnestingFunctionCallExpression(FunctionUtils
-                    .getFunctionInfo(fi), args);
+            UnnestingFunctionCallExpression ufce = new UnnestingFunctionCallExpression(
+                    FunctionUtils.getFunctionInfo(fi), args);
             ufce.setReturnsUniqueValues(AsterixBuiltinFunctions.returnsUniqueValues(fi));
             f = ufce;
         } else {
@@ -590,9 +598,9 @@
         sel1.getInputs().add(new MutableObject<ILogicalOperator>(pThen.first));
 
         Pair<ILogicalOperator, LogicalVariable> pElse = ifexpr.getElseExpr().accept(this, nestedSource);
-        AbstractFunctionCallExpression notVarCond = new ScalarFunctionCallExpression(FunctionUtils
-                .getFunctionInfo(AlgebricksBuiltinFunctions.NOT), new MutableObject<ILogicalExpression>(
-                new VariableReferenceExpression(varCond)));
+        AbstractFunctionCallExpression notVarCond = new ScalarFunctionCallExpression(
+                FunctionUtils.getFunctionInfo(AlgebricksBuiltinFunctions.NOT), new MutableObject<ILogicalExpression>(
+                        new VariableReferenceExpression(varCond)));
         SelectOperator sel2 = new SelectOperator(new MutableObject<ILogicalExpression>(notVarCond));
         sel2.getInputs().add(new MutableObject<ILogicalOperator>(pElse.first));
 
@@ -605,10 +613,10 @@
         sp.getInputs().add(opCondRef);
 
         LogicalVariable resV = context.newVar();
-        AbstractFunctionCallExpression concatNonNull = new ScalarFunctionCallExpression(FunctionUtils
-                .getFunctionInfo(AsterixBuiltinFunctions.CONCAT_NON_NULL), new MutableObject<ILogicalExpression>(
-                new VariableReferenceExpression(pThen.second)), new MutableObject<ILogicalExpression>(
-                new VariableReferenceExpression(pElse.second)));
+        AbstractFunctionCallExpression concatNonNull = new ScalarFunctionCallExpression(
+                FunctionUtils.getFunctionInfo(AsterixBuiltinFunctions.CONCAT_NON_NULL),
+                new MutableObject<ILogicalExpression>(new VariableReferenceExpression(pThen.second)),
+                new MutableObject<ILogicalExpression>(new VariableReferenceExpression(pElse.second)));
         AssignOperator a = new AssignOperator(resV, new MutableObject<ILogicalExpression>(concatNonNull));
         a.getInputs().add(new MutableObject<ILogicalOperator>(sp));
 
@@ -716,9 +724,8 @@
             OrderModifier m = modifIter.next();
             OrderOperator.IOrder comp = (m == OrderModifier.ASC) ? OrderOperator.ASC_ORDER : OrderOperator.DESC_ORDER;
             ord.getOrderExpressions()
-                    .add(
-                            new Pair<IOrder, Mutable<ILogicalExpression>>(comp, new MutableObject<ILogicalExpression>(
-                                    p.first)));
+                    .add(new Pair<IOrder, Mutable<ILogicalExpression>>(comp, new MutableObject<ILogicalExpression>(
+                            p.first)));
             topOp = p.second;
         }
         ord.getInputs().add(topOp);
@@ -772,8 +779,8 @@
         } else { // EVERY
             List<Mutable<ILogicalExpression>> satExprList = new ArrayList<Mutable<ILogicalExpression>>(1);
             satExprList.add(new MutableObject<ILogicalExpression>(eo2.first));
-            s = new SelectOperator(new MutableObject<ILogicalExpression>(new ScalarFunctionCallExpression(FunctionUtils
-                    .getFunctionInfo(AlgebricksBuiltinFunctions.NOT), satExprList)));
+            s = new SelectOperator(new MutableObject<ILogicalExpression>(new ScalarFunctionCallExpression(
+                    FunctionUtils.getFunctionInfo(AlgebricksBuiltinFunctions.NOT), satExprList)));
             s.getInputs().add(eo2.second);
             fAgg = AsterixBuiltinFunctions.makeAggregateFunctionExpression(AsterixBuiltinFunctions.EMPTY_STREAM,
                     new ArrayList<Mutable<ILogicalExpression>>());
@@ -794,8 +801,8 @@
     @Override
     public Pair<ILogicalOperator, LogicalVariable> visitRecordConstructor(RecordConstructor rc,
             Mutable<ILogicalOperator> tupSource) throws AsterixException {
-        AbstractFunctionCallExpression f = new ScalarFunctionCallExpression(FunctionUtils
-                .getFunctionInfo(AsterixBuiltinFunctions.OPEN_RECORD_CONSTRUCTOR));
+        AbstractFunctionCallExpression f = new ScalarFunctionCallExpression(
+                FunctionUtils.getFunctionInfo(AsterixBuiltinFunctions.OPEN_RECORD_CONSTRUCTOR));
         LogicalVariable v1 = context.newVar();
         AssignOperator a = new AssignOperator(v1, new MutableObject<ILogicalExpression>(f));
         Mutable<ILogicalOperator> topOp = tupSource;
@@ -839,8 +846,8 @@
         if (u.getSign() == Sign.POSITIVE) {
             a = new AssignOperator(v1, new MutableObject<ILogicalExpression>(eo.first));
         } else {
-            AbstractFunctionCallExpression m = new ScalarFunctionCallExpression(FunctionUtils
-                    .getFunctionInfo(AsterixBuiltinFunctions.NUMERIC_UNARY_MINUS));
+            AbstractFunctionCallExpression m = new ScalarFunctionCallExpression(
+                    FunctionUtils.getFunctionInfo(AsterixBuiltinFunctions.NUMERIC_UNARY_MINUS));
             m.getArguments().add(new MutableObject<ILogicalExpression>(eo.first));
             a = new AssignOperator(v1, new MutableObject<ILogicalExpression>(m));
         }
@@ -1263,8 +1270,8 @@
     private ILogicalExpression makeUnnestExpression(ILogicalExpression expr) {
         switch (expr.getExpressionTag()) {
             case VARIABLE: {
-                return new UnnestingFunctionCallExpression(FunctionUtils
-                        .getFunctionInfo(AsterixBuiltinFunctions.SCAN_COLLECTION),
+                return new UnnestingFunctionCallExpression(
+                        FunctionUtils.getFunctionInfo(AsterixBuiltinFunctions.SCAN_COLLECTION),
                         new MutableObject<ILogicalExpression>(expr));
             }
             case FUNCTION_CALL: {
@@ -1272,8 +1279,8 @@
                 if (fce.getKind() == FunctionKind.UNNEST) {
                     return expr;
                 } else {
-                    return new UnnestingFunctionCallExpression(FunctionUtils
-                            .getFunctionInfo(AsterixBuiltinFunctions.SCAN_COLLECTION),
+                    return new UnnestingFunctionCallExpression(
+                            FunctionUtils.getFunctionInfo(AsterixBuiltinFunctions.SCAN_COLLECTION),
                             new MutableObject<ILogicalExpression>(expr));
                 }
             }
diff --git a/asterix-app/src/test/resources/logging.properties b/asterix-app/src/test/resources/logging.properties
index deb88307..4cf5545 100644
--- a/asterix-app/src/test/resources/logging.properties
+++ b/asterix-app/src/test/resources/logging.properties
@@ -61,5 +61,5 @@
 # messages:
 
 #edu.uci.ics.asterix.level = FINE
-#edu.uci.ics.algebricks.level = FINE
-#edu.uci.ics.hyracks.level = INFO
+edu.uci.ics.algebricks.level = FINE
+edu.uci.ics.hyracks.level = INFO
diff --git a/asterix-app/src/test/resources/runtimets/ignore.txt b/asterix-app/src/test/resources/runtimets/ignore.txt
index 2b456ca..5ba8ee7 100644
--- a/asterix-app/src/test/resources/runtimets/ignore.txt
+++ b/asterix-app/src/test/resources/runtimets/ignore.txt
@@ -15,4 +15,4 @@
 fuzzyjoin
 failure/q1_pricing_summary_report_failure.aql
 open-closed
-dml/load-from-hdfs.aql
+dml/load-from-hdfs.aql
\ No newline at end of file
diff --git a/asterix-app/src/test/resources/runtimets/queries/dml/insert-into-loaded-dataset-with-index_01.aql b/asterix-app/src/test/resources/runtimets/queries/dml/insert-into-loaded-dataset-with-index_01.aql
index 01e1845..7a3cce2 100644
--- a/asterix-app/src/test/resources/runtimets/queries/dml/insert-into-loaded-dataset-with-index_01.aql
+++ b/asterix-app/src/test/resources/runtimets/queries/dml/insert-into-loaded-dataset-with-index_01.aql
@@ -24,7 +24,7 @@
 let $z:=3
 return {
 	"l_orderkey": $x,
-	"l_partkey": $y,
+	"l_linenumber": $y,
 	"l_suppkey": $z
 }
 );
@@ -35,7 +35,7 @@
 let $z:=4
 return {
 	"l_orderkey": $x,
-	"l_partkey": $y,
+	"l_linenumber": $y,
 	"l_suppkey": $z
 }
 );
diff --git a/asterix-app/src/test/resources/runtimets/queries/dml/insert-into-loaded-dataset-with-index_02.aql b/asterix-app/src/test/resources/runtimets/queries/dml/insert-into-loaded-dataset-with-index_02.aql
index 2212eb9..04a5043 100644
--- a/asterix-app/src/test/resources/runtimets/queries/dml/insert-into-loaded-dataset-with-index_02.aql
+++ b/asterix-app/src/test/resources/runtimets/queries/dml/insert-into-loaded-dataset-with-index_02.aql
@@ -49,7 +49,7 @@
 	where $l.l_orderkey<10
 	return {
 		"l_orderkey": $l.l_orderkey,
-		"l_partkey": $l.l_linenumber,
+		"l_linenumber": $l.l_linenumber,
 		"l_suppkey": $l.l_partkey
 	}
 );
diff --git a/asterix-app/src/test/resources/runtimets/queries/dml/insert-into-loaded-dataset_01.aql b/asterix-app/src/test/resources/runtimets/queries/dml/insert-into-loaded-dataset_01.aql
index ac41ec7..1f211e8 100644
--- a/asterix-app/src/test/resources/runtimets/queries/dml/insert-into-loaded-dataset_01.aql
+++ b/asterix-app/src/test/resources/runtimets/queries/dml/insert-into-loaded-dataset_01.aql
@@ -21,7 +21,7 @@
 let $z:=3
 return {
 	"l_orderkey": $x,
-	"l_partkey": $y,
+	"l_linenumber": $y,
 	"l_suppkey": $z
 }
 );
@@ -32,7 +32,7 @@
 let $z:=4
 return {
 	"l_orderkey": $x,
-	"l_partkey": $y,
+	"l_linenumber": $y,
 	"l_suppkey": $z
 }
 );
diff --git a/asterix-app/src/test/resources/runtimets/queries/dml/insert.aql b/asterix-app/src/test/resources/runtimets/queries/dml/insert.aql
index 7c38a64..2b4622f 100644
--- a/asterix-app/src/test/resources/runtimets/queries/dml/insert.aql
+++ b/asterix-app/src/test/resources/runtimets/queries/dml/insert.aql
@@ -47,7 +47,7 @@
 	where $l.l_orderkey<10
 	return {
 		"l_orderkey": $l.l_orderkey,
-		"l_partkey": $l.l_linenumber,
+		"l_linenumber": $l.l_linenumber,
 		"l_suppkey": $l.l_partkey
 	}
 );
diff --git a/asterix-app/src/test/resources/runtimets/queries/dml/insert_less_nc.aql b/asterix-app/src/test/resources/runtimets/queries/dml/insert_less_nc.aql
index 0fc544d..7e133ae 100644
--- a/asterix-app/src/test/resources/runtimets/queries/dml/insert_less_nc.aql
+++ b/asterix-app/src/test/resources/runtimets/queries/dml/insert_less_nc.aql
@@ -46,7 +46,7 @@
 	where $l.l_orderkey<1000
 	return {
 		"l_orderkey": $l.l_orderkey,
-		"l_partkey": $l.l_linenumber,
+		"l_linenumber": $l.l_linenumber,
 		"l_suppkey": $l.l_partkey
 	}
 );
diff --git a/asterix-app/src/test/resources/runtimets/queries/dml/opentype-c2o.aql b/asterix-app/src/test/resources/runtimets/queries/dml/opentype-c2o.aql
new file mode 100644
index 0000000..ef6cd3a
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/dml/opentype-c2o.aql
@@ -0,0 +1,48 @@
+drop dataverse testdv2 if exists;
+create dataverse testdv2;
+use dataverse testdv2;
+
+create type testtype as closed {
+  id: string,
+  name: string,
+  hobby: string?
+}
+
+create type testtype2 as open {
+  id: string
+}
+
+create dataset testds(testtype) partitioned by key id;
+
+create dataset testds2(testtype2) partitioned by key id;
+ 
+insert into dataset testds (
+{ "id": "001", "name": "Person Three", "hobby": "music"}
+);
+
+insert into dataset testds (
+{ "id": "002", "name": "Person One", "hobby": "sports"}
+);
+
+insert into dataset testds (
+{ "id": "003", "name": "Person Two", "hobby": "movie"}
+);
+
+insert into dataset testds (
+{ "id": "004", "name": "Person Four", "hobby": "swimming"}
+);
+ 
+insert into dataset testds (
+{ "id": "005", "name": "Person Five"}
+); 
+ 
+insert into dataset testds2 (
+ for $d in dataset("testds") 
+	return $d
+);
+
+write output to nc1:"rttest/dml_opentype-c2o.adm";
+
+for $d in dataset("testds2") 
+order by $d.id
+return $d
diff --git a/asterix-app/src/test/resources/runtimets/queries/dml/opentype-closed-optional.aql b/asterix-app/src/test/resources/runtimets/queries/dml/opentype-closed-optional.aql
new file mode 100644
index 0000000..cc3b622
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/dml/opentype-closed-optional.aql
@@ -0,0 +1,24 @@
+drop dataverse testdv2 if exists;
+create dataverse testdv2;
+use dataverse testdv2;
+
+create type testtype as closed {
+  name: string ?,
+  id: string
+}
+
+create dataset testds(testtype) partitioned by key id;
+
+insert into dataset testds (
+{ "id": "001", "name": "Person One"}
+);
+
+insert into dataset testds (
+{ "id": "002"}
+);
+
+write output to nc1:"rttest/dml_opentype-closed-optional.adm";
+
+for $d in dataset("testds") 
+order by $d.id
+return $d
diff --git a/asterix-app/src/test/resources/runtimets/queries/dml/opentype-insert.aql b/asterix-app/src/test/resources/runtimets/queries/dml/opentype-insert.aql
index 0c71f54..19dd1d6 100644
--- a/asterix-app/src/test/resources/runtimets/queries/dml/opentype-insert.aql
+++ b/asterix-app/src/test/resources/runtimets/queries/dml/opentype-insert.aql
@@ -1,3 +1,4 @@
+drop dataverse testdv2 if exists;
 create dataverse testdv2;
 use dataverse testdv2;
 
diff --git a/asterix-app/src/test/resources/runtimets/queries/dml/opentype-noexpand.aql b/asterix-app/src/test/resources/runtimets/queries/dml/opentype-noexpand.aql
new file mode 100644
index 0000000..f8b4880
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/dml/opentype-noexpand.aql
@@ -0,0 +1,24 @@
+drop dataverse testdv2 if exists;
+create dataverse testdv2;
+use dataverse testdv2;
+
+create type testtype as open {
+  name: string ?,
+  id: string
+}
+
+create dataset testds(testtype) partitioned by key id;
+
+insert into dataset testds (
+{ "id": "001", "name": "Person One"}
+);
+
+insert into dataset testds (
+{ "id": "002"}
+);
+
+write output to nc1:"rttest/dml_opentype-noexpand.adm";
+
+for $d in dataset("testds") 
+order by $d.id
+return $d
diff --git a/asterix-app/src/test/resources/runtimets/queries/dml/opentype-o2c.aql b/asterix-app/src/test/resources/runtimets/queries/dml/opentype-o2c.aql
new file mode 100644
index 0000000..d5c2195
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/dml/opentype-o2c.aql
@@ -0,0 +1,45 @@
+drop dataverse testdv2 if exists;
+create dataverse testdv2;
+use dataverse testdv2;
+
+create type testtype as open {
+  id: string,
+  name: string
+}
+
+create type testtype2 as closed {
+  id: string,
+  name: string,
+  hobby: string?
+}
+
+create dataset testds(testtype) partitioned by key id;
+
+create dataset testds2(testtype2) partitioned by key id;
+ 
+insert into dataset testds (
+{ "id": "001", "name": "Person Three", "hobby": "music"}
+);
+
+insert into dataset testds (
+{ "id": "002", "name": "Person Three", "hobby": "football"}
+);
+
+insert into dataset testds (
+{ "id": "003", "name": "Person Three", "hobby": "movie"}
+);
+
+insert into dataset testds (
+{ "id": "004", "name": "Person Three", "hobby": "swimming"}
+);
+
+insert into dataset testds2 (
+ for $d in dataset("testds") 
+	return $d
+);
+
+write output to nc1:"rttest/dml_opentype-o2c.adm";
+
+for $d in dataset("testds2") 
+order by $d.id
+return $d
diff --git a/asterix-app/src/test/resources/runtimets/queries/dml/opentype-o2o.aql b/asterix-app/src/test/resources/runtimets/queries/dml/opentype-o2o.aql
new file mode 100644
index 0000000..6793180
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/queries/dml/opentype-o2o.aql
@@ -0,0 +1,45 @@
+drop dataverse testdv2 if exists;
+create dataverse testdv2;
+use dataverse testdv2;
+
+create type testtype as open {
+  name: string,
+  id: string
+}
+
+create type testtype2 as open {
+  id: string,
+  name: string,
+  hobby: string
+}
+
+create dataset testds(testtype) partitioned by key id;
+
+create dataset testds2(testtype2) partitioned by key id;
+ 
+insert into dataset testds (
+{ "name": "Person One",  "id": "001", "hobby": "music"}
+);
+
+insert into dataset testds (
+{ "name": "Person Two", "id": "002",  "hobby": "football", "city":"irvine"}
+);
+
+insert into dataset testds (
+{ "name": "Person Three",  "id": "003",  "hobby": "movie"}
+);
+
+insert into dataset testds (
+{ "name": "Person Four", "id": "004",  "hobby": "swimming", "phone":"102-304-506"}
+);
+
+insert into dataset testds2 (
+ for $d in dataset("testds") 
+	return $d
+);
+
+write output to nc1:"rttest/dml_opentype-o2o.adm";
+
+for $d in dataset("testds2") 
+order by $d.id
+return $d
diff --git a/asterix-app/src/test/resources/runtimets/queries/failure/insert.aql b/asterix-app/src/test/resources/runtimets/queries/failure/insert.aql
index 3f1daad..a0ce842 100644
--- a/asterix-app/src/test/resources/runtimets/queries/failure/insert.aql
+++ b/asterix-app/src/test/resources/runtimets/queries/failure/insert.aql
@@ -47,7 +47,7 @@
 	where $l.l_orderkey<1000
 	return {
 		"l_orderkey": $l.l_orderkey,
-		"l_partkey": $l.l_linenumber,
+		"l_linenumber": $l.l_linenumber,
 		"l_suppkey": $l.l_partkey
 	}
 );
@@ -58,7 +58,7 @@
 	die after 1000
 	return {
 		"l_orderkey": $l.l_orderkey,
-		"l_partkey": $l.l_linenumber,
+		"l_linenumber": $l.l_linenumber,
 		"l_suppkey": $l.l_partkey
 	}
 );
diff --git a/asterix-app/src/test/resources/runtimets/queries/open-closed/heterog-list01.aql b/asterix-app/src/test/resources/runtimets/queries/open-closed/heterog-list01.aql
index c54d020..d1edc3d 100644
--- a/asterix-app/src/test/resources/runtimets/queries/open-closed/heterog-list01.aql
+++ b/asterix-app/src/test/resources/runtimets/queries/open-closed/heterog-list01.aql
@@ -15,7 +15,7 @@
 descrpt:string
 }
 
-create type TestType as {
+create type TestType as closed {
 id:int32,
 description:string,
 name:string,
@@ -30,3 +30,9 @@
 "name":"Cake",
 "batters":[{"id":345,"descprt":"Regular"},{"id":445,"descprt":"Chocolate"}] }
 );
+
+write output to nc1:"rttest/open-closed_heteror-list01.adm";
+
+for $d in dataset("T1") 
+return $d
+
diff --git a/asterix-app/src/test/resources/runtimets/queries/open-closed/open-closed-01.aql b/asterix-app/src/test/resources/runtimets/queries/open-closed/open-closed-01.aql
index 0b6d0dd..6373569 100644
--- a/asterix-app/src/test/resources/runtimets/queries/open-closed/open-closed-01.aql
+++ b/asterix-app/src/test/resources/runtimets/queries/open-closed/open-closed-01.aql
@@ -18,7 +18,9 @@
 
 create dataset testds(testType) partitioned by key id;
 
+
 insert into dataset testds({"id": 123, "name": "John Doe", "hobbies": {{ "scuba", "music" }} }
 );
 
+write output to nc1:"rttest/open-closed_open-closed-01.adm";
 for $l in dataset("testds") return $l
diff --git a/asterix-app/src/test/resources/runtimets/queries/open-closed/open-closed-02.aql b/asterix-app/src/test/resources/runtimets/queries/open-closed/open-closed-02.aql
index 6963bee..c037cc8 100644
--- a/asterix-app/src/test/resources/runtimets/queries/open-closed/open-closed-02.aql
+++ b/asterix-app/src/test/resources/runtimets/queries/open-closed/open-closed-02.aql
@@ -19,6 +19,7 @@
 
 create dataset testds(testType) partitioned by key id;
 
-insert into dataset testds({"id": 123, "name": "John Doe", "hobbies": {{ "scuba", "music" }} });
+insert into dataset testds({"id": 123);
 
+write output to nc1:"rttest/open-closed_open-closed-02.adm";
 for $l in dataset("testds") return $l
diff --git a/asterix-app/src/test/resources/runtimets/results/dml/opentype-c2o.adm b/asterix-app/src/test/resources/runtimets/results/dml/opentype-c2o.adm
new file mode 100644
index 0000000..834bcef
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/results/dml/opentype-c2o.adm
@@ -0,0 +1,4 @@
+{ "id": "001", "name": "Person Three", "hobby": "music" }
+{ "id": "002", "name": "Person One", "hobby": "sports" }
+{ "id": "003", "name": "Person Two", "hobby": "movie" }
+{ "id": "004", "name": "Person Four", "hobby": "swimming" }
diff --git a/asterix-app/src/test/resources/runtimets/results/dml/opentype-closed-optional.adm b/asterix-app/src/test/resources/runtimets/results/dml/opentype-closed-optional.adm
new file mode 100644
index 0000000..78e8d93
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/results/dml/opentype-closed-optional.adm
@@ -0,0 +1,2 @@
+{ "name": "Person One", "id": "001" }
+{ "name": null, "id": "002" }
diff --git a/asterix-app/src/test/resources/runtimets/results/dml/opentype-noexpand.adm b/asterix-app/src/test/resources/runtimets/results/dml/opentype-noexpand.adm
new file mode 100644
index 0000000..78e8d93
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/results/dml/opentype-noexpand.adm
@@ -0,0 +1,2 @@
+{ "name": "Person One", "id": "001" }
+{ "name": null, "id": "002" }
diff --git a/asterix-app/src/test/resources/runtimets/results/dml/opentype-o2c.adm b/asterix-app/src/test/resources/runtimets/results/dml/opentype-o2c.adm
new file mode 100644
index 0000000..0c1e6d3
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/results/dml/opentype-o2c.adm
@@ -0,0 +1,4 @@
+{ "id": "001", "name": "Person Three", "hobby": "music" }
+{ "id": "002", "name": "Person Three", "hobby": "football" }
+{ "id": "003", "name": "Person Three", "hobby": "movie" }
+{ "id": "004", "name": "Person Three", "hobby": "swimming" }
diff --git a/asterix-app/src/test/resources/runtimets/results/dml/opentype-o2o.adm b/asterix-app/src/test/resources/runtimets/results/dml/opentype-o2o.adm
new file mode 100644
index 0000000..3b25366
--- /dev/null
+++ b/asterix-app/src/test/resources/runtimets/results/dml/opentype-o2o.adm
@@ -0,0 +1,4 @@
+{ "id": "001", "name": "Person One", "hobby": "music" }
+{ "id": "002", "name": "Person Two", "hobby": "football", "city": "irvine" }
+{ "id": "003", "name": "Person Three", "hobby": "movie" }
+{ "id": "004", "name": "Person Four", "hobby": "swimming", "phone": "102-304-506" }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ARecordPrinterFactory.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ARecordPrinterFactory.java
index 184a810..fafe320 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ARecordPrinterFactory.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ARecordPrinterFactory.java
@@ -123,7 +123,7 @@
                                     fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b,
                                             fieldOffsets[fieldNumber], tag, false);
                                     fieldPrinters[fieldNumber].print(b, fieldOffsets[fieldNumber] - 1,
-                                            fieldOffsets[fieldNumber], ps);
+                                            fieldValueLength, ps);
                                 }
                             } else {
                                 tag = fieldTypes[fieldNumber].getTypeTag();
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/functions/AsterixBuiltinFunctions.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/functions/AsterixBuiltinFunctions.java
index 5f8334d..9aeac2b 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/functions/AsterixBuiltinFunctions.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/functions/AsterixBuiltinFunctions.java
@@ -22,6 +22,7 @@
 import edu.uci.ics.asterix.om.typecomputer.impl.ARectangleTypeComputer;
 import edu.uci.ics.asterix.om.typecomputer.impl.AStringTypeComputer;
 import edu.uci.ics.asterix.om.typecomputer.impl.BinaryBooleanOrNullFunctionTypeComputer;
+import edu.uci.ics.asterix.om.typecomputer.impl.CastRecordResultTypeComputer;
 import edu.uci.ics.asterix.om.typecomputer.impl.ClosedRecordConstructorResultType;
 import edu.uci.ics.asterix.om.typecomputer.impl.FieldAccessByIndexResultType;
 import edu.uci.ics.asterix.om.typecomputer.impl.InjectFailureTypeComputer;
@@ -81,7 +82,6 @@
         SI
     }
 
-    private final static Map<AsterixFunction, IFunctionInfo> asterixFunctionToInfo = new HashMap<AsterixFunction, IFunctionInfo>();
     private final static Map<FunctionIdentifier, IFunctionInfo> asterixFunctionIdToInfo = new HashMap<FunctionIdentifier, IFunctionInfo>();
 
     // it is supposed to be an identity mapping
@@ -334,6 +334,8 @@
             true);
     public final static FunctionIdentifier INJECT_FAILURE = new FunctionIdentifier(FunctionConstants.ASTERIX_NS,
             "inject-failure", 2, true);
+    public final static FunctionIdentifier CAST_RECORD = new FunctionIdentifier(FunctionConstants.ASTERIX_NS,
+            "cast-record", 1, true);
 
     public static final FunctionIdentifier EQ = AlgebricksBuiltinFunctions.EQ;
     public static final FunctionIdentifier LE = AlgebricksBuiltinFunctions.LE;
@@ -351,13 +353,13 @@
         IFunctionInfo finfo = asterixFunctionIdToInfo.get(fid);
         if (finfo == null) {
             finfo = new AsterixFunctionInfo(fid, fid.isBuiltin());
-            if(fid.isBuiltin()){
+            if (fid.isBuiltin()) {
                 asterixFunctionIdToInfo.put(fid, finfo);
             }
         }
         return finfo;
     }
-    
+
     public static AsterixFunctionInfo lookupFunction(FunctionIdentifier fid) {
         return (AsterixFunctionInfo) asterixFunctionIdToInfo.get(fid);
     }
@@ -542,6 +544,7 @@
         add(SWITCH_CASE, NonTaggedSwitchCaseComputer.INSTANCE);
         add(REG_EXP, ABooleanTypeComputer.INSTANCE);
         add(INJECT_FAILURE, InjectFailureTypeComputer.INSTANCE);
+        add(CAST_RECORD, CastRecordResultTypeComputer.INSTANCE);
 
         add(TID, AInt32TypeComputer.INSTANCE);
         add(TIME_CONSTRUCTOR, OptionalATimeTypeComputer.INSTANCE);
@@ -676,8 +679,6 @@
         IFunctionInfo finfo = getAsterixFunctionInfo(fi);
         return finfo == null ? null : finfo.getFunctionIdentifier();
     }
-    
-    
 
     public static AggregateFunctionCallExpression makeAggregateFunctionExpression(FunctionIdentifier fi,
             List<Mutable<ILogicalExpression>> args) {
@@ -697,7 +698,7 @@
 
     public static boolean isAggregateFunctionSerializable(FunctionIdentifier fi) {
         IFunctionInfo finfo = getAsterixFunctionInfo(fi);
-        return aggregateToSerializableAggregate.get(finfo) != null ;
+        return aggregateToSerializableAggregate.get(finfo) != null;
     }
 
     public static AggregateFunctionCallExpression makeSerializableAggregateFunctionExpression(FunctionIdentifier fi,
@@ -706,8 +707,9 @@
         IFunctionInfo finfo = getAsterixFunctionInfo(fi);
         IFunctionInfo serializableFinfo = aggregateToSerializableAggregate.get(finfo);
         if (serializableFinfo == null)
-            throw new IllegalStateException("no serializable implementation for aggregate function " + serializableFinfo);
-       
+            throw new IllegalStateException("no serializable implementation for aggregate function "
+                    + serializableFinfo);
+
         IFunctionInfo fiLocal = aggregateToLocalAggregate.get(serializableFinfo);
         IFunctionInfo fiGlobal = aggregateToGlobalAggregate.get(serializableFinfo);
 
@@ -735,7 +737,8 @@
         builtinFunctionsSet.put(functionInfo, functionInfo);
         funTypeComputer.put(functionInfo, typeComputer);
         asterixFunctionIdToInfo.put(fi, functionInfo);
-       // AsterixFunction asterixFunction = new AsterixFunction(fi.getName(), fi.getArity());
+        // AsterixFunction asterixFunction = new AsterixFunction(fi.getName(),
+        // fi.getArity());
     }
 
     private static void addAgg(FunctionIdentifier fi) {
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/base/TypeComputerUtilities.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/base/TypeComputerUtilities.java
index abd4d66..e8786ec 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/base/TypeComputerUtilities.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/base/TypeComputerUtilities.java
@@ -21,13 +21,14 @@
         return openType;
     }
 
-    public static boolean setRequiredType(AbstractFunctionCallExpression expr, ARecordType requiredRecordType) {
+    public static boolean setRequiredAndInputTypes(AbstractFunctionCallExpression expr, ARecordType requiredRecordType,
+            ARecordType inputRecordType) {
         boolean changed = false;
         Object opaqueParameter = expr.getOpaqueParameters();
         if (opaqueParameter == null) {
-            opaqueParameter = requiredRecordType;
-            Object[] opaqueParameters = new Object[1];
-            opaqueParameters[0] = opaqueParameter;
+            Object[] opaqueParameters = new Object[2];
+            opaqueParameters[0] = requiredRecordType;
+            opaqueParameters[1] = inputRecordType;
             expr.setOpaqueParameters(opaqueParameters);
             changed = true;
         }
@@ -43,4 +44,13 @@
             return null;
     }
 
+    public static ARecordType getInputType(AbstractFunctionCallExpression expr) {
+        Object[] type = expr.getOpaqueParameters();
+        if (type != null) {
+            ARecordType recordType = (ARecordType) type[1];
+            return recordType;
+        } else
+            return null;
+    }
+
 }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/CastRecordResultTypeComputer.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/CastRecordResultTypeComputer.java
new file mode 100644
index 0000000..40560fa
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/CastRecordResultTypeComputer.java
@@ -0,0 +1,22 @@
+package edu.uci.ics.asterix.om.typecomputer.impl;
+
+import edu.uci.ics.asterix.om.typecomputer.base.IResultTypeComputer;
+import edu.uci.ics.asterix.om.typecomputer.base.TypeComputerUtilities;
+import edu.uci.ics.asterix.om.types.IAType;
+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.expressions.IVariableTypeEnvironment;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.metadata.IMetadataProvider;
+
+public class CastRecordResultTypeComputer implements IResultTypeComputer {
+
+    public static final CastRecordResultTypeComputer INSTANCE = new CastRecordResultTypeComputer();
+
+    @Override
+    public IAType computeType(ILogicalExpression expression, IVariableTypeEnvironment env,
+            IMetadataProvider<?, ?> metadataProvider) throws AlgebricksException {
+        ScalarFunctionCallExpression funcExpr = (ScalarFunctionCallExpression) expression;
+        return TypeComputerUtilities.getRequiredType(funcExpr);
+    }
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/ClosedRecordConstructorResultType.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/ClosedRecordConstructorResultType.java
index 3fcf1c8..6a17a46 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/ClosedRecordConstructorResultType.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/ClosedRecordConstructorResultType.java
@@ -7,6 +7,7 @@
 import edu.uci.ics.asterix.om.base.AString;
 import edu.uci.ics.asterix.om.constants.AsterixConstantValue;
 import edu.uci.ics.asterix.om.typecomputer.base.IResultTypeComputer;
+import edu.uci.ics.asterix.om.typecomputer.base.TypeComputerUtilities;
 import edu.uci.ics.asterix.om.types.ARecordType;
 import edu.uci.ics.asterix.om.types.IAType;
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
@@ -26,6 +27,14 @@
     public IAType computeType(ILogicalExpression expression, IVariableTypeEnvironment env,
             IMetadataProvider<?, ?> metadataProvider) throws AlgebricksException {
         AbstractFunctionCallExpression f = (AbstractFunctionCallExpression) expression;
+
+        /**
+         * if type has been top-down propagated, use the enforced type
+         */
+        ARecordType type = TypeComputerUtilities.getRequiredType(f);
+        if (type != null)
+            return type;
+
         int n = f.getArguments().size() / 2;
         String[] fieldNames = new String[n];
         IAType[] fieldTypes = new IAType[n];
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/RecordConstructorResultType.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/RecordConstructorResultType.java
index 8327f30..d20f43b 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/RecordConstructorResultType.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/RecordConstructorResultType.java
@@ -7,6 +7,7 @@
 import edu.uci.ics.asterix.om.base.AString;
 import edu.uci.ics.asterix.om.constants.AsterixConstantValue;
 import edu.uci.ics.asterix.om.typecomputer.base.IResultTypeComputer;
+import edu.uci.ics.asterix.om.typecomputer.base.TypeComputerUtilities;
 import edu.uci.ics.asterix.om.types.ARecordType;
 import edu.uci.ics.asterix.om.types.IAType;
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
@@ -32,6 +33,9 @@
     public IAType computeType(ILogicalExpression expression, IVariableTypeEnvironment env,
             IMetadataProvider<?, ?> metadataProvider) throws AlgebricksException {
         AbstractFunctionCallExpression f = (AbstractFunctionCallExpression) expression;
+        IAType reqType = TypeComputerUtilities.getRequiredType(f);
+        if (reqType != null)
+            return reqType;
         int n = f.getArguments().size() / 2;
         String[] fieldNames = new String[n];
         IAType[] fieldTypes = new IAType[n];
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CastRecordDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CastRecordDescriptor.java
new file mode 100644
index 0000000..f7d3887
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CastRecordDescriptor.java
@@ -0,0 +1,71 @@
+package edu.uci.ics.asterix.runtime.evaluators.functions;
+
+import java.io.DataOutput;
+import java.io.IOException;
+
+import edu.uci.ics.asterix.common.functions.FunctionConstants;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.runtime.evaluators.base.AbstractScalarFunctionDynamicDescriptor;
+import edu.uci.ics.asterix.runtime.util.ARecordAccessor;
+import edu.uci.ics.asterix.runtime.util.ARecordCaster;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import edu.uci.ics.hyracks.algebricks.runtime.base.IEvaluator;
+import edu.uci.ics.hyracks.algebricks.runtime.base.IEvaluatorFactory;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ArrayBackedValueStorage;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.IDataOutputProvider;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+
+public class CastRecordDescriptor extends AbstractScalarFunctionDynamicDescriptor {
+
+    protected static final FunctionIdentifier FID_CAST = new FunctionIdentifier(FunctionConstants.ASTERIX_NS,
+            "cast-record", 1, true);
+
+    private static final long serialVersionUID = 1L;
+    private ARecordType reqType;
+    private ARecordType inputType;
+
+    public void reset(ARecordType reqType, ARecordType inputType) {
+        this.reqType = reqType;
+        this.inputType = inputType;
+    }
+
+    @Override
+    public FunctionIdentifier getIdentifier() {
+        return FID_CAST;
+    }
+
+    @Override
+    public IEvaluatorFactory createEvaluatorFactory(final IEvaluatorFactory[] args) {
+        final IEvaluatorFactory recordEvalFactory = args[0];
+
+        return new IEvaluatorFactory() {
+            private static final long serialVersionUID = 1L;
+
+            @Override
+            public IEvaluator createEvaluator(final IDataOutputProvider output) throws AlgebricksException {
+                final DataOutput out = output.getDataOutput();
+                final ArrayBackedValueStorage recordBuffer = new ArrayBackedValueStorage();
+                final IEvaluator recEvaluator = recordEvalFactory.createEvaluator(recordBuffer);
+
+                return new IEvaluator() {
+                    final ARecordAccessor recAccessor = new ARecordAccessor(inputType);
+                    final ARecordCaster caster = new ARecordCaster();
+
+                    @Override
+                    public void evaluate(IFrameTupleReference tuple) throws AlgebricksException {
+                        try {
+                            recordBuffer.reset();
+                            recEvaluator.evaluate(tuple);
+                            recAccessor.reset(recordBuffer.getBytes(), recordBuffer.getStartIndex(),
+                                    recordBuffer.getLength());
+                            caster.castRecord(recAccessor, reqType, out);
+                        } catch (IOException ioe) {
+                            throw new AlgebricksException(ioe);
+                        }
+                    }
+                };
+            }
+        };
+    }
+}
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/formats/NonTaggedDataFormat.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/formats/NonTaggedDataFormat.java
index 467cd03..dd984d9 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/formats/NonTaggedDataFormat.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/formats/NonTaggedDataFormat.java
@@ -32,6 +32,7 @@
 import edu.uci.ics.asterix.om.functions.FunctionManagerHolder;
 import edu.uci.ics.asterix.om.functions.IFunctionDescriptor;
 import edu.uci.ics.asterix.om.functions.IFunctionManager;
+import edu.uci.ics.asterix.om.typecomputer.base.TypeComputerUtilities;
 import edu.uci.ics.asterix.om.types.AOrderedListType;
 import edu.uci.ics.asterix.om.types.ARecordType;
 import edu.uci.ics.asterix.om.types.ATypeTag;
@@ -76,6 +77,7 @@
 import edu.uci.ics.asterix.runtime.evaluators.constructors.ATimeConstructorDescriptor;
 import edu.uci.ics.asterix.runtime.evaluators.functions.AndDescriptor;
 import edu.uci.ics.asterix.runtime.evaluators.functions.AnyCollectionMemberDescriptor;
+import edu.uci.ics.asterix.runtime.evaluators.functions.CastRecordDescriptor;
 import edu.uci.ics.asterix.runtime.evaluators.functions.ClosedRecordConstructorDescriptor;
 import edu.uci.ics.asterix.runtime.evaluators.functions.ContainsDescriptor;
 import edu.uci.ics.asterix.runtime.evaluators.functions.CountHashedGramTokensDescriptor;
@@ -311,6 +313,7 @@
         temp.add(new SwitchCaseDescriptor());
         temp.add(new RegExpDescriptor());
         temp.add(new InjectFailureDescriptor());
+        temp.add(new CastRecordDescriptor());
 
         IFunctionManager mgr = new FunctionManagerImpl();
         for (IFunctionDescriptor fd : temp) {
@@ -362,8 +365,8 @@
                 } catch (HyracksDataException e) {
                     throw new AlgebricksException(e);
                 }
-                IEvaluatorFactory fldIndexEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs.getBytes(), abvs
-                        .getLength()));
+                IEvaluatorFactory fldIndexEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs.getBytes(),
+                        abvs.getLength()));
                 IEvaluatorFactory evalFactory = new FieldAccessByIndexEvalFactory(recordEvalFactory,
                         fldIndexEvalFactory, recType);
                 return evalFactory;
@@ -388,8 +391,8 @@
         } catch (HyracksDataException e) {
             throw new AlgebricksException(e);
         }
-        IEvaluatorFactory dimensionEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs1.getBytes(), abvs1
-                .getLength()));
+        IEvaluatorFactory dimensionEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs1.getBytes(),
+                abvs1.getLength()));
 
         for (int i = 0; i < numOfFields; i++) {
             ArrayBackedValueStorage abvs2 = new ArrayBackedValueStorage();
@@ -400,8 +403,8 @@
             } catch (HyracksDataException e) {
                 throw new AlgebricksException(e);
             }
-            IEvaluatorFactory coordinateEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs2.getBytes(), abvs2
-                    .getLength()));
+            IEvaluatorFactory coordinateEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs2.getBytes(),
+                    abvs2.getLength()));
 
             evalFactories[i] = new CreateMBREvalFactory(evalFactory, dimensionEvalFactory, coordinateEvalFactory);
         }
@@ -427,8 +430,8 @@
                 } catch (HyracksDataException e) {
                     throw new AlgebricksException(e);
                 }
-                IEvaluatorFactory fldIndexEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs.getBytes(), abvs
-                        .getLength()));
+                IEvaluatorFactory fldIndexEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs.getBytes(),
+                        abvs.getLength()));
                 IEvaluatorFactory evalFactory = new FieldAccessByIndexEvalFactory(recordEvalFactory,
                         fldIndexEvalFactory, recType);
                 IFunctionInfo finfoAccess = AsterixBuiltinFunctions
@@ -469,10 +472,15 @@
                 ((ListifyAggregateDescriptor) fd).reset(new AOrderedListType(itemType, null));
             }
         }
+        if (fd.getIdentifier().equals(AsterixBuiltinFunctions.CAST_RECORD)) {
+            ARecordType rt = (ARecordType) TypeComputerUtilities.getRequiredType((AbstractFunctionCallExpression) expr);
+            ARecordType it = (ARecordType) TypeComputerUtilities.getInputType((AbstractFunctionCallExpression) expr);
+            ((CastRecordDescriptor) fd).reset(rt, it);
+        }
         if (fd.getIdentifier().equals(AsterixBuiltinFunctions.OPEN_RECORD_CONSTRUCTOR)) {
             ARecordType rt = (ARecordType) context.getType(expr);
-            ((OpenRecordConstructorDescriptor) fd).reset(rt, computeOpenFields((AbstractFunctionCallExpression) expr,
-                    rt));
+            ((OpenRecordConstructorDescriptor) fd).reset(rt,
+                    computeOpenFields((AbstractFunctionCallExpression) expr, rt));
         }
         if (fd.getIdentifier().equals(AsterixBuiltinFunctions.CLOSED_RECORD_CONSTRUCTOR)) {
             ((ClosedRecordConstructorDescriptor) fd).reset((ARecordType) context.getType(expr));
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ARecordAccessor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ARecordAccessor.java
new file mode 100644
index 0000000..9a048f5
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ARecordAccessor.java
@@ -0,0 +1,269 @@
+package edu.uci.ics.asterix.runtime.util;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import edu.uci.ics.asterix.dataflow.data.nontagged.AqlNullWriterFactory;
+import edu.uci.ics.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
+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.EnumDeserializer;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
+import edu.uci.ics.hyracks.api.dataflow.value.INullWriter;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
+
+public class ARecordAccessor implements IValueReference {
+
+    private List<SimpleValueReference> fieldNames = new ArrayList<SimpleValueReference>();
+    private List<SimpleValueReference> fieldTypeTags = new ArrayList<SimpleValueReference>();
+    private List<SimpleValueReference> fieldValues = new ArrayList<SimpleValueReference>();
+
+    private byte[] typeBuffer = new byte[32768];
+    private ResetableByteArrayOutputStream typeBos = new ResetableByteArrayOutputStream();
+    private DataOutputStream typeDos = new DataOutputStream(typeBos);
+
+    private byte[] dataBuffer = new byte[32768];
+    private ResetableByteArrayOutputStream dataBos = new ResetableByteArrayOutputStream();
+    private DataOutputStream dataDos = new DataOutputStream(dataBos);
+
+    private int closedPartTypeInfoSize = 0;
+    private ARecordType inputRecType;
+
+    private int numberOfSchemaFields;
+    private int offsetArrayOffset;
+    private int[] fieldOffsets;
+    private int fieldCursor = -1;
+    private ATypeTag typeTag;
+    private SimpleValueReference nullReference = new SimpleValueReference();
+
+    private byte[] data;
+    private int start;
+    private int len;
+
+    public ARecordAccessor(ARecordType inputType) {
+        this.inputRecType = inputType;
+        IAType[] fieldTypes = inputType.getFieldTypes();
+        String[] fieldNameStrs = inputType.getFieldNames();
+        numberOfSchemaFields = fieldTypes.length;
+
+        // initialize the buffer for closed parts(fieldName bytes+ type bytes) +
+        // constant(null bytes)
+        typeBos.setByteArray(typeBuffer, 0);
+        try {
+            for (int i = 0; i < numberOfSchemaFields; i++) {
+                ATypeTag ftypeTag = fieldTypes[i].getTypeTag();
+
+                // add type tag Reference
+                int tagStart = typeBos.size();
+                typeDos.writeByte(ftypeTag.serialize());
+                int tagEnd = typeBos.size();
+                SimpleValueReference typeTagReference = new SimpleValueReference();
+                typeTagReference.reset(typeBuffer, tagStart, tagEnd - tagStart);
+                fieldTypeTags.add(typeTagReference);
+
+                // add type name Reference (including a astring type tag)
+                int nameStart = typeBos.size();
+                typeDos.writeByte(ATypeTag.STRING.serialize());
+                typeDos.writeUTF(fieldNameStrs[i]);
+                int nameEnd = typeBos.size();
+                SimpleValueReference typeNameReference = new SimpleValueReference();
+                typeNameReference.reset(typeBuffer, nameStart, nameEnd - nameStart);
+                fieldNames.add(typeNameReference);
+            }
+
+            // initialize a constant: null value bytes reference
+            int nullFieldStart = typeBos.size();
+            INullWriter nullWriter = AqlNullWriterFactory.INSTANCE.createNullWriter();
+            nullWriter.writeNull(typeDos);
+            int nullFieldEnd = typeBos.size();
+            nullReference.reset(typeBuffer, nullFieldStart, nullFieldEnd - nullFieldStart);
+        } catch (IOException e) {
+            throw new IllegalStateException(e);
+        }
+        closedPartTypeInfoSize = typeBos.size();
+        fieldOffsets = new int[numberOfSchemaFields];
+    }
+
+    private void reset() {
+        typeBos.setByteArray(typeBuffer, closedPartTypeInfoSize);
+        dataBos.setByteArray(dataBuffer, 0);
+        fieldCursor = -1;
+    }
+
+    public void reset(byte[] b, int start, int len) {
+        // clear the previous states
+        reset();
+        this.data = b;
+        this.start = start;
+        this.len = len;
+
+        boolean isExpanded = false;
+        int openPartOffset = 0;
+        int s = start;
+        int recordOffset = s;
+        if (inputRecType == null) {
+            openPartOffset = s + AInt32SerializerDeserializer.getInt(b, s + 6);
+            s += 8;
+            isExpanded = true;
+        } else {
+            if (inputRecType.isOpen()) {
+                isExpanded = b[s + 5] == 1 ? true : false;
+                if (isExpanded) {
+                    openPartOffset = s + AInt32SerializerDeserializer.getInt(b, s + 6);
+                    s += 10;
+                } else
+                    s += 6;
+            } else
+                s += 5;
+        }
+        try {
+            if (numberOfSchemaFields > 0) {
+                s += 4;
+                int nullBitMapOffset = 0;
+                boolean hasNullableFields = NonTaggedFormatUtil.hasNullableField(inputRecType);
+                if (hasNullableFields) {
+                    nullBitMapOffset = s;
+                    offsetArrayOffset = s
+                            + (this.numberOfSchemaFields % 8 == 0 ? numberOfSchemaFields / 8
+                                    : numberOfSchemaFields / 8 + 1);
+                } else {
+                    offsetArrayOffset = s;
+                }
+                for (int i = 0; i < numberOfSchemaFields; i++) {
+                    fieldOffsets[i] = AInt32SerializerDeserializer.getInt(b, offsetArrayOffset) + recordOffset;
+                    offsetArrayOffset += 4;
+                }
+                for (int fieldNumber = 0; fieldNumber < numberOfSchemaFields; fieldNumber++) {
+                    next();
+                    if (hasNullableFields) {
+                        byte b1 = b[nullBitMapOffset + fieldNumber / 8];
+                        int p = 1 << (7 - (fieldNumber % 8));
+                        if ((b1 & p) == 0) {
+                            // set null value (including type tag inside)
+                            nextFieldValue().reset(nullReference);
+                            continue;
+                        }
+                    }
+                    IAType[] fieldTypes = inputRecType.getFieldTypes();
+                    int fieldValueLength = 0;
+
+                    if (fieldTypes[fieldNumber].getTypeTag() == ATypeTag.UNION) {
+                        if (NonTaggedFormatUtil.isOptionalField((AUnionType) fieldTypes[fieldNumber])) {
+                            typeTag = ((AUnionType) fieldTypes[fieldNumber]).getUnionList()
+                                    .get(NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST).getTypeTag();
+                            fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b, fieldOffsets[fieldNumber],
+                                    typeTag, false);
+                        }
+                    } else {
+                        typeTag = fieldTypes[fieldNumber].getTypeTag();
+                        fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b, fieldOffsets[fieldNumber],
+                                typeTag, false);
+                    }
+                    // set field value (including the type tag)
+                    int fstart = dataBos.size();
+                    dataDos.writeByte(typeTag.serialize());
+                    dataDos.write(b, fieldOffsets[fieldNumber], fieldValueLength);
+                    int fend = dataBos.size();
+                    nextFieldValue().reset(dataBuffer, fstart, fend - fstart);
+                }
+            }
+            if (isExpanded) {
+                int numberOfOpenFields = AInt32SerializerDeserializer.getInt(b, openPartOffset);
+                int fieldOffset = openPartOffset + 4 + (8 * numberOfOpenFields);
+                for (int i = 0; i < numberOfOpenFields; i++) {
+                    next();
+                    // set the field name (including a type tag, which is
+                    // astring)
+                    int fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b, fieldOffset, ATypeTag.STRING,
+                            false);
+                    int fnstart = dataBos.size();
+                    dataDos.writeByte(ATypeTag.STRING.serialize());
+                    dataDos.write(b, fieldOffset, fieldValueLength);
+                    int fnend = dataBos.size();
+                    nextFieldName().reset(dataBuffer, fnstart, fnend - fnstart);
+                    fieldOffset += fieldValueLength;
+
+                    // set the field type tag
+                    nextFieldType().reset(b, fieldOffset, 1);
+                    typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b[fieldOffset]);
+
+                    // set the field value (already including type tag)
+                    fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b, fieldOffset, typeTag, true) + 1;
+                    nextFieldValue().reset(b, fieldOffset, fieldValueLength);
+                    fieldOffset += fieldValueLength;
+                }
+            }
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+    }
+
+    private void next() {
+        fieldCursor++;
+    }
+
+    private SimpleValueReference nextFieldName() {
+        if (fieldCursor < fieldNames.size()) {
+            return fieldNames.get(fieldCursor);
+        } else {
+            SimpleValueReference fieldNameReference = new SimpleValueReference();
+            fieldNames.add(fieldNameReference);
+            return fieldNameReference;
+        }
+    }
+
+    private SimpleValueReference nextFieldType() {
+        if (fieldCursor < fieldTypeTags.size()) {
+            return fieldTypeTags.get(fieldCursor);
+        } else {
+            SimpleValueReference fieldTypeReference = new SimpleValueReference();
+            fieldTypeTags.add(fieldTypeReference);
+            return fieldTypeReference;
+        }
+    }
+
+    private SimpleValueReference nextFieldValue() {
+        if (fieldCursor < fieldValues.size()) {
+            return fieldValues.get(fieldCursor);
+        } else {
+            SimpleValueReference fieldValueReference = new SimpleValueReference();
+            fieldValues.add(fieldValueReference);
+            return fieldValueReference;
+        }
+    }
+
+    public int getCursor() {
+        return fieldCursor;
+    }
+
+    public List<SimpleValueReference> getFieldNames() {
+        return fieldNames;
+    }
+
+    public List<SimpleValueReference> getFieldTypeTags() {
+        return fieldTypeTags;
+    }
+
+    public List<SimpleValueReference> getFieldValues() {
+        return fieldValues;
+    }
+
+    @Override
+    public byte[] getBytes() {
+        return data;
+    }
+
+    @Override
+    public int getStartIndex() {
+        return start;
+    }
+
+    @Override
+    public int getLength() {
+        return len;
+    }
+}
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ARecordCaster.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ARecordCaster.java
new file mode 100644
index 0000000..afd11de
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ARecordCaster.java
@@ -0,0 +1,212 @@
+package edu.uci.ics.asterix.runtime.util;
+
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import edu.uci.ics.asterix.builders.RecordBuilder;
+import edu.uci.ics.asterix.dataflow.data.nontagged.AqlNullWriterFactory;
+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.api.dataflow.value.INullWriter;
+
+public class ARecordCaster {
+
+    // describe closed fields in the required type
+    private int[] fieldPermutation;
+
+    // describe fields (open or not) in the input records
+    private boolean[] openFields;
+    private boolean[] optionalFields;
+
+    private List<SimpleValueReference> reqFieldNames = new ArrayList<SimpleValueReference>();
+    private List<SimpleValueReference> reqFieldTypeTags = new ArrayList<SimpleValueReference>();
+    private ARecordType cachedReqType = null;
+
+    private byte[] buffer = new byte[32768];
+    private ResetableByteArrayOutputStream bos = new ResetableByteArrayOutputStream();
+    private DataOutputStream dos = new DataOutputStream(bos);
+
+    private RecordBuilder recBuilder = new RecordBuilder();
+    private SimpleValueReference nullReference = new SimpleValueReference();
+    private SimpleValueReference nullTypeTag = new SimpleValueReference();
+
+    public ARecordCaster() {
+        try {
+            bos.setByteArray(buffer, 0);
+            int start = bos.size();
+            INullWriter nullWriter = AqlNullWriterFactory.INSTANCE.createNullWriter();
+            nullWriter.writeNull(dos);
+            int end = bos.size();
+            nullReference.reset(buffer, start, end - start);
+            start = bos.size();
+            dos.write(ATypeTag.NULL.serialize());
+            end = bos.size();
+            nullTypeTag.reset(buffer, start, end);
+        } catch (IOException e) {
+            throw new IllegalStateException(e);
+        }
+    }
+
+    public void castRecord(ARecordAccessor recordAccessor, ARecordType reqType, DataOutput output) throws IOException {
+        List<SimpleValueReference> fieldNames = recordAccessor.getFieldNames();
+        List<SimpleValueReference> fieldTypeTags = recordAccessor.getFieldTypeTags();
+        List<SimpleValueReference> fieldValues = recordAccessor.getFieldValues();
+
+        if (openFields == null || fieldNames.size() > openFields.length) {
+            openFields = new boolean[fieldNames.size()];
+        }
+        if (cachedReqType == null || !reqType.equals(cachedReqType)) {
+            loadRequiredType(reqType);
+        }
+
+        // clear the previous states
+        reset();
+        matchClosedPart(fieldNames, fieldTypeTags, fieldValues);
+        writeOutput(fieldNames, fieldTypeTags, fieldValues, output);
+    }
+
+    private void reset() {
+        for (int i = 0; i < openFields.length; i++)
+            openFields[i] = true;
+        for (int i = 0; i < fieldPermutation.length; i++)
+            fieldPermutation[i] = -1;
+    }
+
+    private void loadRequiredType(ARecordType reqType) throws IOException {
+        reqFieldNames.clear();
+        reqFieldTypeTags.clear();
+
+        cachedReqType = reqType;
+        int numSchemaFields = reqType.getFieldTypes().length;
+        IAType[] fieldTypes = reqType.getFieldTypes();
+        String[] fieldNames = reqType.getFieldNames();
+        fieldPermutation = new int[numSchemaFields];
+        optionalFields = new boolean[numSchemaFields];
+        for (int i = 0; i < optionalFields.length; i++)
+            optionalFields[i] = false;
+
+        bos.setByteArray(buffer, nullReference.getStartIndex() + nullReference.getLength());
+        for (int i = 0; i < numSchemaFields; i++) {
+            ATypeTag ftypeTag = fieldTypes[i].getTypeTag();
+            String fname = fieldNames[i];
+
+            // add type tag pointable
+            if (fieldTypes[i].getTypeTag() == ATypeTag.UNION
+                    && NonTaggedFormatUtil.isOptionalField((AUnionType) fieldTypes[i])) {
+                // optional field: add the embedded non-null type tag
+                ftypeTag = ((AUnionType) fieldTypes[i]).getUnionList()
+                        .get(NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST).getTypeTag();
+                optionalFields[i] = true;
+            }
+            int tagStart = bos.size();
+            dos.writeByte(ftypeTag.serialize());
+            int tagEnd = bos.size();
+            SimpleValueReference typeTagPointable = new SimpleValueReference();
+            typeTagPointable.reset(buffer, tagStart, tagEnd - tagStart);
+            reqFieldTypeTags.add(typeTagPointable);
+
+            // add type name pointable (including a string type tag)
+            int nameStart = bos.size();
+            dos.write(ATypeTag.STRING.serialize());
+            dos.writeUTF(fname);
+            int nameEnd = bos.size();
+            SimpleValueReference typeNamePointable = new SimpleValueReference();
+            typeNamePointable.reset(buffer, nameStart, nameEnd - nameStart);
+            reqFieldNames.add(typeNamePointable);
+        }
+    }
+
+    private void matchClosedPart(List<SimpleValueReference> fieldNames, List<SimpleValueReference> fieldTypeTags,
+            List<SimpleValueReference> fieldValues) {
+        // forward match: match from actual to required
+        boolean matched = false;
+        for (int i = 0; i < fieldNames.size(); i++) {
+            SimpleValueReference fieldName = fieldNames.get(i);
+            SimpleValueReference fieldTypeTag = fieldTypeTags.get(i);
+            matched = false;
+            for (int j = 0; j < reqFieldNames.size(); j++) {
+                SimpleValueReference reqFieldName = reqFieldNames.get(j);
+                SimpleValueReference reqFieldTypeTag = reqFieldTypeTags.get(j);
+                if (fieldName.equals(reqFieldName) && fieldTypeTag.equals(reqFieldTypeTag)) {
+                    fieldPermutation[j] = i;
+                    openFields[i] = false;
+                    matched = true;
+                    break;
+                }
+            }
+            if (matched)
+                continue;
+            // the input has extra fields
+            if (!cachedReqType.isOpen())
+                throw new IllegalStateException("type mismatch: including extra closed fields");
+        }
+
+        // backward match: match from required to actual
+        for (int i = 0; i < reqFieldNames.size(); i++) {
+            SimpleValueReference reqFieldName = reqFieldNames.get(i);
+            SimpleValueReference reqFieldTypeTag = reqFieldTypeTags.get(i);
+            matched = false;
+            for (int j = 0; j < fieldNames.size(); j++) {
+                SimpleValueReference fieldName = fieldNames.get(j);
+                SimpleValueReference fieldTypeTag = fieldTypeTags.get(j);
+                if (fieldName.equals(reqFieldName)) {
+                    if (fieldTypeTag.equals(reqFieldTypeTag)) {
+                        matched = true;
+                        break;
+                    }
+
+                    // match the null type of optional field
+                    if (optionalFields[i] && fieldTypeTag.equals(nullTypeTag)) {
+                        matched = true;
+                        break;
+                    }
+                }
+            }
+            if (matched)
+                continue;
+
+            IAType t = cachedReqType.getFieldTypes()[i];
+            if (t.getTypeTag() == ATypeTag.UNION && NonTaggedFormatUtil.isOptionalField((AUnionType) t)) {
+                // add a null field into the end of field name list and type
+                // list
+                fieldNames.add(reqFieldName);
+                fieldTypeTags.add(reqFieldTypeTag);
+                fieldValues.add(nullReference);
+                fieldPermutation[i] = fieldNames.size() - 1;
+            } else {
+                // no matched field in the input for a required closed field
+                throw new IllegalStateException("type mismatch: miss a required closed field");
+            }
+        }
+    }
+
+    private void writeOutput(List<SimpleValueReference> fieldNames, List<SimpleValueReference> fieldTypeTags,
+            List<SimpleValueReference> fieldValues, DataOutput output) throws IOException {
+        // reset the states of the record builder
+        recBuilder.reset(cachedReqType);
+        recBuilder.init();
+
+        // write the closed part
+        for (int i = 0; i < fieldPermutation.length; i++) {
+            int pos = fieldPermutation[i];
+            SimpleValueReference field = fieldValues.get(pos);
+            recBuilder.addField(i, field);
+        }
+
+        // write the open part
+        for (int i = 0; i < openFields.length; i++) {
+            if (openFields[i]) {
+                SimpleValueReference name = fieldNames.get(i);
+                SimpleValueReference field = fieldValues.get(i);
+                recBuilder.addField(name, field);
+            }
+        }
+        recBuilder.write(output, true);
+    }
+}
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ResetableByteArrayInputStream.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ResetableByteArrayInputStream.java
new file mode 100644
index 0000000..7f8f0e9
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ResetableByteArrayInputStream.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright 2009-2010 by The Regents of the University of California
+ * Licensed 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 from
+ * 
+ *     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 edu.uci.ics.asterix.runtime.util;
+
+import java.io.ByteArrayInputStream;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public class ResetableByteArrayInputStream extends ByteArrayInputStream {
+    private static final Logger LOGGER = Logger.getLogger(ResetableByteArrayInputStream.class.getName());
+
+    private byte[] data;
+    private int position;
+
+    public ResetableByteArrayInputStream(byte[] data) {
+        super(data);
+    }
+
+    public void setByteArray(byte[] data, int position) {
+        this.data = data;
+        this.position = position;
+    }
+
+    @Override
+    public int read() {
+        int remaining = data.length - position;
+        int value = remaining > 0 ? (data[position++] & 0xff) : -1;
+        if (LOGGER.isLoggable(Level.FINEST)) {
+            LOGGER.finest("read(): value: " + value + " remaining: " + remaining + " position: " + position);
+        }
+        return value;
+    }
+
+    @Override
+    public int read(byte[] bytes, int offset, int length) {
+        int remaining = data.length - position;
+        if (LOGGER.isLoggable(Level.FINEST)) {
+            LOGGER.finest("read(bytes[], int, int): remaining: " + remaining + " offset: " + offset + " length: "
+                    + length + " position: " + position);
+        }
+        if (remaining == 0) {
+            return -1;
+        }
+        int l = Math.min(length, remaining);
+        System.arraycopy(data, position, bytes, offset, l);
+        position += l;
+        return l;
+    }
+}
\ No newline at end of file
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ResetableByteArrayOutputStream.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ResetableByteArrayOutputStream.java
new file mode 100644
index 0000000..2958314
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ResetableByteArrayOutputStream.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright 2009-2010 by The Regents of the University of California
+ * Licensed 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 from
+ * 
+ *     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 edu.uci.ics.asterix.runtime.util;
+
+import java.io.ByteArrayOutputStream;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public class ResetableByteArrayOutputStream extends ByteArrayOutputStream {
+    private static final Logger LOGGER = Logger.getLogger(ResetableByteArrayOutputStream.class.getName());
+
+    private byte[] data;
+    private int position;
+
+    public ResetableByteArrayOutputStream() {
+    }
+
+    public void setByteArray(byte[] data, int position) {
+        this.data = data;
+        this.position = position;
+    }
+
+    @Override
+    public void write(int b) {
+        int remaining = data.length - position;
+        if (position + 1 > data.length - 1)
+            throw new IndexOutOfBoundsException();
+        data[position] = (byte) b;
+        position++;
+        if (LOGGER.isLoggable(Level.FINEST)) {
+            LOGGER.finest("write(): value: " + b + " remaining: " + remaining + " position: " + position);
+        }
+    }
+
+    @Override
+    public void write(byte[] bytes, int offset, int length) {
+        if (LOGGER.isLoggable(Level.FINEST)) {
+            LOGGER.finest("write(bytes[], int, int) offset: " + offset + " length: " + length + " position: "
+                    + position);
+        }
+        if (position + length > data.length - 1)
+            throw new IndexOutOfBoundsException();
+        System.arraycopy(bytes, offset, data, position, length);
+        position += length;
+    }
+    
+    @Override
+    public int size(){
+        return position;
+    }
+}
\ No newline at end of file
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/SimpleValueReference.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/SimpleValueReference.java
new file mode 100644
index 0000000..bb54624
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/SimpleValueReference.java
@@ -0,0 +1,55 @@
+package edu.uci.ics.asterix.runtime.util;
+
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
+
+public class SimpleValueReference implements IValueReference {
+
+    private byte[] data;
+    private int start;
+    private int len;
+
+    public void reset(byte[] data, int start, int len) {
+        this.data = data;
+        this.start = start;
+        this.len = len;
+    }
+
+    public void reset(IValueReference ivf) {
+        this.data = ivf.getBytes();
+        this.start = ivf.getStartIndex();
+        this.len = ivf.getLength();
+    }
+
+    @Override
+    public byte[] getBytes() {
+        return data;
+    }
+
+    @Override
+    public int getStartIndex() {
+        return start;
+    }
+
+    @Override
+    public int getLength() {
+        return len;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof IValueReference))
+            return false;
+        IValueReference ivf = (IValueReference) o;
+        byte[] odata = ivf.getBytes();
+        int ostart = ivf.getStartIndex();
+        int olen = ivf.getLength();
+
+        if (len != olen)
+            return false;
+        for (int i = 0; i < len; i++) {
+            if (data[start + i] != odata[ostart + i])
+                return false;
+        }
+        return true;
+    }
+}