diff --git a/asterix-app/src/test/resources/runtimets/only.txt b/asterix-app/src/test/resources/runtimets/only.txt
index e69de29..5b5deb7 100644
--- a/asterix-app/src/test/resources/runtimets/only.txt
+++ b/asterix-app/src/test/resources/runtimets/only.txt
@@ -0,0 +1 @@
+dml/
\ 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/accessors/AFlatValueAccessor.java
similarity index 77%
rename from asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/SimpleValueReference.java
rename to asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/AFlatValueAccessor.java
index e49218a..fe8b656 100644
--- 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/accessors/AFlatValueAccessor.java
@@ -13,28 +13,26 @@
  * limitations under the License.
  */
 
-package edu.uci.ics.asterix.runtime.util;
+package edu.uci.ics.asterix.runtime.accessors;
 
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.runtime.accessors.base.IBinaryAccessor;
+import edu.uci.ics.asterix.runtime.accessors.visitor.IBinaryAccessorVisitor;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
 
-public class SimpleValueReference implements IValueReference {
+public class AFlatValueAccessor implements IBinaryAccessor {
 
     private byte[] data;
     private int start;
     private int len;
 
+    @Override
     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;
@@ -67,4 +65,9 @@
         }
         return true;
     }
+
+    @Override
+    public <R, T> R accept(IBinaryAccessorVisitor<R, T> vistor, T tag) throws AsterixException{
+        return vistor.visit(this, tag);
+    }
 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/AListAccessor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/AListAccessor.java
new file mode 100644
index 0000000..60d6a2e
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/AListAccessor.java
@@ -0,0 +1,38 @@
+package edu.uci.ics.asterix.runtime.accessors;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.runtime.accessors.base.IBinaryAccessor;
+import edu.uci.ics.asterix.runtime.accessors.visitor.IBinaryAccessorVisitor;
+
+public class AListAccessor implements IBinaryAccessor {
+
+    @Override
+    public byte[] getBytes() {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+    @Override
+    public int getLength() {
+        // TODO Auto-generated method stub
+        return 0;
+    }
+
+    @Override
+    public int getStartIndex() {
+        // TODO Auto-generated method stub
+        return 0;
+    }
+
+    @Override
+    public void reset(byte[] b, int start, int len) {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public <R, T> R accept(IBinaryAccessorVisitor<R, T> vistor, T tag) throws AsterixException {
+        return vistor.visit(this, tag);
+    }
+
+}
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/accessors/ARecordAccessor.java
similarity index 85%
rename from asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ARecordAccessor.java
rename to asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/ARecordAccessor.java
index d45b4ac..63bc63e 100644
--- 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/accessors/ARecordAccessor.java
@@ -13,13 +13,14 @@
  * limitations under the License.
  */
 
-package edu.uci.ics.asterix.runtime.util;
+package edu.uci.ics.asterix.runtime.accessors;
 
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
 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;
@@ -28,14 +29,16 @@
 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.asterix.runtime.accessors.base.IBinaryAccessor;
+import edu.uci.ics.asterix.runtime.accessors.visitor.IBinaryAccessorVisitor;
+import edu.uci.ics.asterix.runtime.util.ResettableByteArrayOutputStream;
 import edu.uci.ics.hyracks.api.dataflow.value.INullWriter;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
 
-public class ARecordAccessor implements IValueReference {
+public class ARecordAccessor implements IBinaryAccessor {
 
-    private List<SimpleValueReference> fieldNames = new ArrayList<SimpleValueReference>();
-    private List<SimpleValueReference> fieldTypeTags = new ArrayList<SimpleValueReference>();
-    private List<SimpleValueReference> fieldValues = new ArrayList<SimpleValueReference>();
+    private List<AFlatValueAccessor> fieldNames = new ArrayList<AFlatValueAccessor>();
+    private List<AFlatValueAccessor> fieldTypeTags = new ArrayList<AFlatValueAccessor>();
+    private List<AFlatValueAccessor> fieldValues = new ArrayList<AFlatValueAccessor>();
 
     private byte[] typeBuffer = new byte[32768];
     private ResettableByteArrayOutputStream typeBos = new ResettableByteArrayOutputStream();
@@ -53,7 +56,7 @@
     private int[] fieldOffsets;
     private int fieldCursor = -1;
     private ATypeTag typeTag;
-    private SimpleValueReference nullReference = new SimpleValueReference();
+    private AFlatValueAccessor nullReference = new AFlatValueAccessor();
 
     private byte[] data;
     private int start;
@@ -76,7 +79,7 @@
                 int tagStart = typeBos.size();
                 typeDos.writeByte(ftypeTag.serialize());
                 int tagEnd = typeBos.size();
-                SimpleValueReference typeTagReference = new SimpleValueReference();
+                AFlatValueAccessor typeTagReference = new AFlatValueAccessor();
                 typeTagReference.reset(typeBuffer, tagStart, tagEnd - tagStart);
                 fieldTypeTags.add(typeTagReference);
 
@@ -85,7 +88,7 @@
                 typeDos.writeByte(ATypeTag.STRING.serialize());
                 typeDos.writeUTF(fieldNameStrs[i]);
                 int nameEnd = typeBos.size();
-                SimpleValueReference typeNameReference = new SimpleValueReference();
+                AFlatValueAccessor typeNameReference = new AFlatValueAccessor();
                 typeNameReference.reset(typeBuffer, nameStart, nameEnd - nameStart);
                 fieldNames.add(typeNameReference);
             }
@@ -159,7 +162,8 @@
                         int p = 1 << (7 - (fieldNumber % 8));
                         if ((b1 & p) == 0) {
                             // set null value (including type tag inside)
-                            nextFieldValue().reset(nullReference);
+                            nextFieldValue().reset(nullReference.getBytes(), nullReference.getStartIndex(),
+                                    nullReference.getLength());
                             continue;
                         }
                     }
@@ -221,31 +225,31 @@
         fieldCursor++;
     }
 
-    private SimpleValueReference nextFieldName() {
+    private AFlatValueAccessor nextFieldName() {
         if (fieldCursor < fieldNames.size()) {
             return fieldNames.get(fieldCursor);
         } else {
-            SimpleValueReference fieldNameReference = new SimpleValueReference();
+            AFlatValueAccessor fieldNameReference = new AFlatValueAccessor();
             fieldNames.add(fieldNameReference);
             return fieldNameReference;
         }
     }
 
-    private SimpleValueReference nextFieldType() {
+    private AFlatValueAccessor nextFieldType() {
         if (fieldCursor < fieldTypeTags.size()) {
             return fieldTypeTags.get(fieldCursor);
         } else {
-            SimpleValueReference fieldTypeReference = new SimpleValueReference();
+            AFlatValueAccessor fieldTypeReference = new AFlatValueAccessor();
             fieldTypeTags.add(fieldTypeReference);
             return fieldTypeReference;
         }
     }
 
-    private SimpleValueReference nextFieldValue() {
+    private AFlatValueAccessor nextFieldValue() {
         if (fieldCursor < fieldValues.size()) {
             return fieldValues.get(fieldCursor);
         } else {
-            SimpleValueReference fieldValueReference = new SimpleValueReference();
+            AFlatValueAccessor fieldValueReference = new AFlatValueAccessor();
             fieldValues.add(fieldValueReference);
             return fieldValueReference;
         }
@@ -255,15 +259,15 @@
         return fieldCursor;
     }
 
-    public List<SimpleValueReference> getFieldNames() {
+    public List<AFlatValueAccessor> getFieldNames() {
         return fieldNames;
     }
 
-    public List<SimpleValueReference> getFieldTypeTags() {
+    public List<AFlatValueAccessor> getFieldTypeTags() {
         return fieldTypeTags;
     }
 
-    public List<SimpleValueReference> getFieldValues() {
+    public List<AFlatValueAccessor> getFieldValues() {
         return fieldValues;
     }
 
@@ -281,4 +285,10 @@
     public int getLength() {
         return len;
     }
+
+    @Override
+    public <R, T> R accept(IBinaryAccessorVisitor<R, T> vistor, T tag) throws AsterixException {
+        return vistor.visit(this, tag);
+    }
+
 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/base/IBinaryAccessor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/base/IBinaryAccessor.java
new file mode 100644
index 0000000..37d8b80
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/base/IBinaryAccessor.java
@@ -0,0 +1,27 @@
+/*
+ * 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.accessors.base;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.runtime.accessors.visitor.IBinaryAccessorVisitor;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
+
+public interface IBinaryAccessor extends IValueReference {
+
+    public void reset(byte[] b, int start, int len);
+
+    public <R, T> R accept(IBinaryAccessorVisitor<R, T> vistor, T tag) throws AsterixException;
+}
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/cast/ACastVisitor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/cast/ACastVisitor.java
new file mode 100644
index 0000000..6e874cf
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/cast/ACastVisitor.java
@@ -0,0 +1,44 @@
+package edu.uci.ics.asterix.runtime.accessors.cast;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.runtime.accessors.AFlatValueAccessor;
+import edu.uci.ics.asterix.runtime.accessors.AListAccessor;
+import edu.uci.ics.asterix.runtime.accessors.ARecordAccessor;
+import edu.uci.ics.asterix.runtime.accessors.visitor.IBinaryAccessorVisitor;
+import edu.uci.ics.hyracks.algebricks.common.utils.Triple;
+
+public class ACastVisitor implements IBinaryAccessorVisitor<Void, Triple<ARecordAccessor, IAType, Boolean>> {
+
+    private Map<ARecordAccessor, ARecordCaster> raccessorToCaster = new HashMap<ARecordAccessor, ARecordCaster>();
+
+    @Override
+    public Void visit(AListAccessor accessor, Triple<ARecordAccessor, IAType, Boolean> arg) {
+        return null;
+    }
+
+    @Override
+    public Void visit(ARecordAccessor accessor, Triple<ARecordAccessor, IAType, Boolean> arg) throws AsterixException {
+        ARecordCaster caster = raccessorToCaster.get(accessor);
+        if (caster == null) {
+            caster = new ARecordCaster();
+            raccessorToCaster.put(accessor, caster);
+        }
+        try {
+            caster.castRecord(accessor, arg.first, (ARecordType) arg.second, this);
+        } catch (Exception e) {
+            throw new AsterixException(e);
+        }
+        return null;
+    }
+
+    @Override
+    public Void visit(AFlatValueAccessor accessor, Triple<ARecordAccessor, IAType, Boolean> arg) {
+        return null;
+    }
+
+}
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/accessors/cast/ARecordCaster.java
similarity index 80%
rename from asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ARecordCaster.java
rename to asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/cast/ARecordCaster.java
index 505e769..7b3968c 100644
--- 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/accessors/cast/ARecordCaster.java
@@ -13,7 +13,7 @@
  * limitations under the License.
  */
 
-package edu.uci.ics.asterix.runtime.util;
+package edu.uci.ics.asterix.runtime.accessors.cast;
 
 import java.io.DataOutput;
 import java.io.DataOutputStream;
@@ -28,13 +28,16 @@
 import edu.uci.ics.asterix.om.types.AUnionType;
 import edu.uci.ics.asterix.om.types.IAType;
 import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
+import edu.uci.ics.asterix.runtime.accessors.AFlatValueAccessor;
+import edu.uci.ics.asterix.runtime.accessors.ARecordAccessor;
+import edu.uci.ics.asterix.runtime.util.ResettableByteArrayOutputStream;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
 import edu.uci.ics.hyracks.api.dataflow.value.INullWriter;
 import edu.uci.ics.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
 import edu.uci.ics.hyracks.data.std.primitive.UTF8StringPointable;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
 
-public class ARecordCaster {
+class ARecordCaster {
 
     // describe closed fields in the required type
     private int[] fieldPermutation;
@@ -44,9 +47,9 @@
     private boolean[] openFields;
     private int[] fieldNamesSortedIndex;
 
-    private List<SimpleValueReference> reqFieldNames = new ArrayList<SimpleValueReference>();
+    private List<AFlatValueAccessor> reqFieldNames = new ArrayList<AFlatValueAccessor>();
     private int[] reqFieldNamesSortedIndex;
-    private List<SimpleValueReference> reqFieldTypeTags = new ArrayList<SimpleValueReference>();
+    private List<AFlatValueAccessor> reqFieldTypeTags = new ArrayList<AFlatValueAccessor>();
     private ARecordType cachedReqType = null;
 
     private byte[] buffer = new byte[32768];
@@ -54,13 +57,17 @@
     private DataOutputStream dos = new DataOutputStream(bos);
 
     private RecordBuilder recBuilder = new RecordBuilder();
-    private SimpleValueReference nullReference = new SimpleValueReference();
-    private SimpleValueReference nullTypeTag = new SimpleValueReference();
+    private AFlatValueAccessor nullReference = new AFlatValueAccessor();
+    private AFlatValueAccessor nullTypeTag = new AFlatValueAccessor();
 
     private int numInputFields = 0;
     private IBinaryComparator fieldNameComparator = PointableBinaryComparatorFactory.of(UTF8StringPointable.FACTORY)
             .createBinaryComparator();
 
+    private byte[] outputBuffer = new byte[32768];
+    private ResettableByteArrayOutputStream outputBos = new ResettableByteArrayOutputStream();
+    private DataOutputStream outputDos = new DataOutputStream(outputBos);
+
     public ARecordCaster() {
         try {
             bos.setByteArray(buffer, 0);
@@ -78,10 +85,11 @@
         }
     }
 
-    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();
+    public void castRecord(ARecordAccessor recordAccessor, ARecordAccessor resultAccessor, ARecordType reqType,
+            ACastVisitor visitor) throws IOException {
+        List<AFlatValueAccessor> fieldNames = recordAccessor.getFieldNames();
+        List<AFlatValueAccessor> fieldTypeTags = recordAccessor.getFieldTypeTags();
+        List<AFlatValueAccessor> fieldValues = recordAccessor.getFieldValues();
         numInputFields = recordAccessor.getCursor() + 1;
 
         if (openFields == null || numInputFields > openFields.length) {
@@ -95,7 +103,8 @@
         // clear the previous states
         reset();
         matchClosedPart(fieldNames, fieldTypeTags, fieldValues);
-        writeOutput(fieldNames, fieldTypeTags, fieldValues, output);
+        writeOutput(fieldNames, fieldTypeTags, fieldValues, outputDos, visitor);
+        resultAccessor.reset(outputBuffer, 0, outputBos.size());
     }
 
     private void reset() {
@@ -105,6 +114,7 @@
             fieldPermutation[i] = -1;
         for (int i = 0; i < numInputFields; i++)
             fieldNamesSortedIndex[i] = i;
+        outputBos.setByteArray(outputBuffer, 0);
     }
 
     private void loadRequiredType(ARecordType reqType) throws IOException {
@@ -136,7 +146,7 @@
             int tagStart = bos.size();
             dos.writeByte(ftypeTag.serialize());
             int tagEnd = bos.size();
-            SimpleValueReference typeTagPointable = new SimpleValueReference();
+            AFlatValueAccessor typeTagPointable = new AFlatValueAccessor();
             typeTagPointable.reset(buffer, tagStart, tagEnd - tagStart);
             reqFieldTypeTags.add(typeTagPointable);
 
@@ -145,7 +155,7 @@
             dos.write(ATypeTag.STRING.serialize());
             dos.writeUTF(fname);
             int nameEnd = bos.size();
-            SimpleValueReference typeNamePointable = new SimpleValueReference();
+            AFlatValueAccessor typeNamePointable = new AFlatValueAccessor();
             typeNamePointable.reset(buffer, nameStart, nameEnd - nameStart);
             reqFieldNames.add(typeNamePointable);
         }
@@ -157,8 +167,8 @@
         quickSort(reqFieldNamesSortedIndex, reqFieldNames, 0, reqFieldNamesSortedIndex.length - 1);
     }
 
-    private void matchClosedPart(List<SimpleValueReference> fieldNames, List<SimpleValueReference> fieldTypeTags,
-            List<SimpleValueReference> fieldValues) {
+    private void matchClosedPart(List<AFlatValueAccessor> fieldNames, List<AFlatValueAccessor> fieldTypeTags,
+            List<AFlatValueAccessor> fieldValues) {
         // sort-merge based match
         quickSort(fieldNamesSortedIndex, fieldNames, 0, numInputFields - 1);
         int fnStart = 0;
@@ -168,8 +178,8 @@
             int reqFnPos = reqFieldNamesSortedIndex[reqFnStart];
             int c = compare(fieldNames.get(fnPos), reqFieldNames.get(reqFnPos));
             if (c == 0) {
-                SimpleValueReference fieldTypeTag = fieldTypeTags.get(fnPos);
-                SimpleValueReference reqFieldTypeTag = reqFieldTypeTags.get(reqFnPos);
+                AFlatValueAccessor fieldTypeTag = fieldTypeTags.get(fnPos);
+                AFlatValueAccessor reqFieldTypeTag = reqFieldTypeTags.get(reqFnPos);
                 if (fieldTypeTag.equals(reqFieldTypeTag) || (
                 // match the null type of optional field
                         optionalFields[reqFnPos] && fieldTypeTag.equals(nullTypeTag))) {
@@ -203,8 +213,8 @@
         }
     }
 
-    private void writeOutput(List<SimpleValueReference> fieldNames, List<SimpleValueReference> fieldTypeTags,
-            List<SimpleValueReference> fieldValues, DataOutput output) throws IOException {
+    private void writeOutput(List<AFlatValueAccessor> fieldNames, List<AFlatValueAccessor> fieldTypeTags,
+            List<AFlatValueAccessor> fieldValues, DataOutput output, ACastVisitor visitor) throws IOException {
         // reset the states of the record builder
         recBuilder.reset(cachedReqType);
         recBuilder.init();
@@ -212,7 +222,7 @@
         // write the closed part
         for (int i = 0; i < fieldPermutation.length; i++) {
             int pos = fieldPermutation[i];
-            SimpleValueReference field;
+            AFlatValueAccessor field;
             if (pos >= 0) {
                 field = fieldValues.get(pos);
             } else {
@@ -224,15 +234,15 @@
         // write the open part
         for (int i = 0; i < numInputFields; i++) {
             if (openFields[i]) {
-                SimpleValueReference name = fieldNames.get(i);
-                SimpleValueReference field = fieldValues.get(i);
+                AFlatValueAccessor name = fieldNames.get(i);
+                AFlatValueAccessor field = fieldValues.get(i);
                 recBuilder.addField(name, field);
             }
         }
         recBuilder.write(output, true);
     }
 
-    private void quickSort(int[] index, List<SimpleValueReference> names, int start, int end) {
+    private void quickSort(int[] index, List<AFlatValueAccessor> names, int start, int end) {
         if (end <= start)
             return;
         int i = partition(index, names, start, end);
@@ -240,7 +250,7 @@
         quickSort(index, names, i + 1, end);
     }
 
-    private int partition(int[] index, List<SimpleValueReference> names, int left, int right) {
+    private int partition(int[] index, List<AFlatValueAccessor> names, int left, int right) {
         int i = left - 1;
         int j = right;
         while (true) {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/visitor/IBinaryAccessorVisitor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/visitor/IBinaryAccessorVisitor.java
new file mode 100644
index 0000000..44b8722
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/visitor/IBinaryAccessorVisitor.java
@@ -0,0 +1,15 @@
+package edu.uci.ics.asterix.runtime.accessors.visitor;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.runtime.accessors.AFlatValueAccessor;
+import edu.uci.ics.asterix.runtime.accessors.AListAccessor;
+import edu.uci.ics.asterix.runtime.accessors.ARecordAccessor;
+
+public interface IBinaryAccessorVisitor<R, T> {
+
+    public R visit(AListAccessor accessor, T arg) throws AsterixException;
+
+    public R visit(ARecordAccessor accessor, T arg) throws AsterixException;
+
+    public R visit(AFlatValueAccessor accessor, T arg) throws AsterixException;
+}
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
index 705efab..d193e85 100644
--- 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
@@ -1,16 +1,17 @@
 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.functions.IFunctionDescriptor;
 import edu.uci.ics.asterix.om.functions.IFunctionDescriptorFactory;
 import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.runtime.accessors.ARecordAccessor;
+import edu.uci.ics.asterix.runtime.accessors.cast.ACastVisitor;
 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.common.utils.Triple;
 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;
@@ -57,7 +58,10 @@
 
                 return new IEvaluator() {
                     final ARecordAccessor recAccessor = new ARecordAccessor(inputType);
-                    final ARecordCaster caster = new ARecordCaster();
+                    final ARecordAccessor resultAccessor = new ARecordAccessor(reqType);
+                    final ACastVisitor castVisitor = new ACastVisitor();
+                    final Triple<ARecordAccessor, IAType, Boolean> arg = new Triple<ARecordAccessor, IAType, Boolean>(
+                            resultAccessor, reqType, Boolean.FALSE);
 
                     @Override
                     public void evaluate(IFrameTupleReference tuple) throws AlgebricksException {
@@ -66,8 +70,10 @@
                             recEvaluator.evaluate(tuple);
                             recAccessor.reset(recordBuffer.getBytes(), recordBuffer.getStartIndex(),
                                     recordBuffer.getLength());
-                            caster.castRecord(recAccessor, reqType, out);
-                        } catch (IOException ioe) {
+                            recAccessor.accept(castVisitor, arg);
+                            out.write(resultAccessor.getBytes(), resultAccessor.getStartIndex(),
+                                    resultAccessor.getLength());
+                        } catch (Exception ioe) {
                             throw new AlgebricksException(ioe);
                         }
                     }