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 a8e6961..b6d75bf 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
@@ -184,7 +184,7 @@
             
             @SuppressWarnings("rawtypes")
             ISerializerDeserializer serde = _jobGenCtx.getSerializerDeserializerProvider().getSerializerDeserializer(t);
-            bbis.setByteBuffer(ByteBuffer.wrap(resStore.getBytes(), resStore.getStartIndex(), resStore.getLength()), 0);
+            bbis.setByteBuffer(ByteBuffer.wrap(resStore.getByteArray(), resStore.getStartOffset(), resStore.getLength()), 0);
             IAObject o;
             try {
                 o = (IAObject) serde.deserialize(dis);
diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceDynamicTypeCastRule.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceDynamicTypeCastRule.java
index 4859b09..298bac0 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceDynamicTypeCastRule.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceDynamicTypeCastRule.java
@@ -44,13 +44,27 @@
 import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
 
 /**
- * dynamically cast a constant from its type produced by the originated
- * expression to its required type, in a recursive way
- * it enables:
- * 1. bag-based fields in a record
- * 2. bidirectional cast of a open field and a matched closed field
- * 3. put in null fields when necessary
- * since we have open records, so dynamic cast is needed
+ * dynamically cast a variable from its type to a specified required type, in a
+ * recursive way it enables: 1. bag-based fields in a record, 2. bidirectional
+ * cast of a open field and a matched closed field, and 3. put in null fields
+ * when necessary
+ * 
+ * Here is example: A record { "hobby": {{"music", "coding"}}, "id": "001",
+ * "name": "Person Three"} which confirms to closed type ( id: string, name:
+ * string, hobby: {{string}}? ) can be casted to a open type (id: string ), or
+ * vice versa.
+ * 
+ * However, if input record is a variable, then we don't know its exact field
+ * layout at compile time, for example, records confirming the same type can
+ * different field ordering, different open part.
+ * 
+ * Note that as we can see in the example, the ordering of fields of a record is
+ * not required. Since the open/close part of a record has completely different
+ * underlying memory/storage layout, a cast-record function will change the
+ * layout as specified at runtime.
+ * 
+ * Implementation wise, this rule checks the target dataset type and the input
+ * record type, if the types are different, then enforce a cast function.
  */
 public class IntroduceDynamicTypeCastRule implements IAlgebraicRewriteRule {
 
diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceStaticTypeCastRule.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceStaticTypeCastRule.java
index b91abe7..d6f2670 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceStaticTypeCastRule.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/optimizer/rules/IntroduceStaticTypeCastRule.java
@@ -35,7 +35,7 @@
 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.asterix.runtime.accessors.base.DefaultOpenFieldType;
+import edu.uci.ics.asterix.runtime.pointables.base.DefaultOpenFieldType;
 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;
@@ -56,12 +56,28 @@
 /**
  * statically cast a constant from its type produced by the originated
  * expression to its required type, in a recursive way it enables: 1. bag-based
- * fields in a record 2. bidirectional cast of a open field and a matched closed
- * field 3. put in null fields when necessary It should be fired before the
- * constant folding rule
+ * fields in a record, 2. bidirectional cast of a open field and a matched
+ * closed field, and 3. put in null fields when necessary. It should be fired
+ * before the constant folding rule.
  * 
- * This rule is not responsible for type casting between primitive types Tests
- * open-closed/open-closed-15 and after are not to test this rule
+ * This rule is not responsible for type casting between primitive types.
+ * 
+ * Here is example: A record { "hobby": {{"music", "coding"}}, "id": "001",
+ * "name": "Person Three"} which confirms to closed type ( id: string, name:
+ * string, hobby: {{string}}? ) can be casted to a open type (id: string ), or
+ * vice versa.
+ * 
+ * If the record is a constant, the situation that we are going into insert the
+ * record into a dataset with a different type can be capatured at the compile
+ * time, and type cast is done in the rule.
+ * 
+ * Implementation wise: first, we match the record's type and its target dataset
+ * type to see if it is cast-able; second, if the types are cast-able, we embed the require type to the
+ * original producer expression. If the types are not cast-able, we throw
+ * compile time exceptions.
+ * 
+ * Then, at runtime, the constructors know what to do by checking the required
+ * output type.
  */
 public class IntroduceStaticTypeCastRule implements IAlgebraicRewriteRule {
 
@@ -169,10 +185,14 @@
 
     /**
      * only called when funcExpr is record constructor
-     * @param funcExpr  record constructor function expression
-     * @param requiredListType  required record type
-     * @param inputRecordType 
-     * @param env   type environment
+     * 
+     * @param funcExpr
+     *            record constructor function expression
+     * @param requiredListType
+     *            required record type
+     * @param inputRecordType
+     * @param env
+     *            type environment
      * @throws AlgebricksException
      */
     private void rewriteRecordFuncExpr(ScalarFunctionCallExpression funcExpr, ARecordType requiredRecordType,
@@ -186,10 +206,14 @@
 
     /**
      * only called when funcExpr is list constructor
-     * @param funcExpr  list constructor function expression
-     * @param requiredListType  required list type
-     * @param inputListType 
-     * @param env   type environment
+     * 
+     * @param funcExpr
+     *            list constructor function expression
+     * @param requiredListType
+     *            required list type
+     * @param inputListType
+     * @param env
+     *            type environment
      * @throws AlgebricksException
      */
     private void rewriteListFuncExpr(ScalarFunctionCallExpression funcExpr, AbstractCollectionType requiredListType,
@@ -229,7 +253,7 @@
         int[] fieldPermutation = new int[reqFieldTypes.length];
         boolean[] nullFields = new boolean[reqFieldTypes.length];
         boolean[] openFields = new boolean[inputFieldTypes.length];
-        
+
         Arrays.fill(nullFields, false);
         Arrays.fill(openFields, true);
         Arrays.fill(fieldPermutation, -1);
@@ -295,7 +319,7 @@
             }
             // the input has extra fields
             if (!matched && !reqType.isOpen())
-                throw new AlgebricksException("static type mismatch: including an extra closed field "  + fieldName);
+                throw new AlgebricksException("static type mismatch: including an extra closed field " + fieldName);
         }
 
         // backward match: match from required to actual
diff --git a/asterix-common/src/main/java/edu/uci/ics/asterix/common/utils/UTF8CharSequence.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/utils/UTF8CharSequence.java
index fc5c930..3e03c0a 100644
--- a/asterix-common/src/main/java/edu/uci/ics/asterix/common/utils/UTF8CharSequence.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/utils/UTF8CharSequence.java
@@ -1,7 +1,7 @@
 package edu.uci.ics.asterix.common.utils;
 
+import edu.uci.ics.hyracks.data.std.api.IValueReference;
 import edu.uci.ics.hyracks.data.std.primitive.UTF8StringPointable;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
 import edu.uci.ics.hyracks.dataflow.common.data.util.StringUtils;
 
 public class UTF8CharSequence implements CharSequence {
@@ -48,7 +48,7 @@
         int sStart = start + 2;
         int c = 0;
         int i = 0;
-        byte[] bytes = valueRef.getBytes();
+        byte[] bytes = valueRef.getByteArray();
         while (c < len) {
             buf[i++] = UTF8StringPointable.charAt(bytes, sStart + c);
             c += UTF8StringPointable.charSize(bytes, sStart + c);
@@ -57,7 +57,7 @@
     }
 
     private void resetLength(IValueReference valueRef) {
-        this.len = UTF8StringPointable.getUTFLen(valueRef.getBytes(), start);
+        this.len = UTF8StringPointable.getUTFLen(valueRef.getByteArray(), start);
     }
 
 }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/IAOrderedListBuilder.java b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/IAOrderedListBuilder.java
index 881c302..0ca40a5 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/IAOrderedListBuilder.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/IAOrderedListBuilder.java
@@ -4,7 +4,7 @@
 
 import edu.uci.ics.asterix.om.types.AOrderedListType;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
+import edu.uci.ics.hyracks.data.std.api.IValueReference;
 
 public interface IAOrderedListBuilder {
 
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/IARecordBuilder.java b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/IARecordBuilder.java
index 18699f1..544b532 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/IARecordBuilder.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/IARecordBuilder.java
@@ -19,7 +19,7 @@
 import java.io.IOException;
 
 import edu.uci.ics.asterix.om.types.ARecordType;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
+import edu.uci.ics.hyracks.data.std.api.IValueReference;
 
 /**
  * A record builder helps to construct Asterix Records in a serialized format.
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/IAUnorderedListBuilder.java b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/IAUnorderedListBuilder.java
index f962dfb..50d2038 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/IAUnorderedListBuilder.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/IAUnorderedListBuilder.java
@@ -4,7 +4,7 @@
 
 import edu.uci.ics.asterix.om.types.AUnorderedListType;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
+import edu.uci.ics.hyracks.data.std.api.IValueReference;
 
 public interface IAUnorderedListBuilder {
 
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/OrderedListBuilder.java b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/OrderedListBuilder.java
index dcb0b87..18cec15 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/OrderedListBuilder.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/OrderedListBuilder.java
@@ -10,7 +10,7 @@
 import edu.uci.ics.asterix.om.types.ATypeTag;
 import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
+import edu.uci.ics.hyracks.data.std.api.IValueReference;
 
 public class OrderedListBuilder implements IAOrderedListBuilder {
 
@@ -60,10 +60,10 @@
             this.offsets.add((short) outputStream.size());
         if (itemTypeTag == ATypeTag.ANY) {
             this.numberOfItems++;
-            this.outputStream.write(item.getBytes(), item.getStartIndex(), item.getLength());
-        } else if (item.getBytes()[0] != serNullTypeTag) {
+            this.outputStream.write(item.getByteArray(), item.getStartOffset(), item.getLength());
+        } else if (item.getByteArray()[0] != serNullTypeTag) {
             this.numberOfItems++;
-            this.outputStream.write(item.getBytes(), item.getStartIndex() + 1, item.getLength() - 1);
+            this.outputStream.write(item.getByteArray(), item.getStartOffset() + 1, item.getLength() - 1);
         }
     }
 
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/RecordBuilder.java b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/RecordBuilder.java
index fb8d2e4..03f0e20 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/RecordBuilder.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/RecordBuilder.java
@@ -11,8 +11,8 @@
 import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryHashFunction;
 import edu.uci.ics.hyracks.data.std.accessors.PointableBinaryHashFunctionFactory;
+import edu.uci.ics.hyracks.data.std.api.IValueReference;
 import edu.uci.ics.hyracks.data.std.primitive.UTF8StringPointable;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
 
 public class RecordBuilder implements IARecordBuilder {
     private int openPartOffset;
@@ -127,9 +127,9 @@
         closedPartOffsets[id] = closedPartOutputStream.size();
         int len = value.getLength() - 1;
         // +1 because we do not store the value tag.
-        closedPartOutputStream.write(value.getBytes(), value.getStartIndex() + 1, len);
+        closedPartOutputStream.write(value.getByteArray(), value.getStartOffset() + 1, len);
         numberOfClosedFields++;
-        if (isNullable && value.getBytes()[0] != SER_NULL_TYPE_TAG) {
+        if (isNullable && value.getByteArray()[0] != SER_NULL_TYPE_TAG) {
             nullBitMap[id / 8] |= (byte) (1 << (7 - (id % 8)));
         }
     }
@@ -142,12 +142,12 @@
             for (int i = 0; i < tempOpenPartOffsets.length; i++)
                 openPartOffsets[i] = tempOpenPartOffsets[i];
         }
-        fieldNameHashCode = utf8HashFunction.hash(name.getBytes(), name.getStartIndex() + 1, name.getLength());
+        fieldNameHashCode = utf8HashFunction.hash(name.getByteArray(), name.getStartOffset() + 1, name.getLength());
         openPartOffsets[this.numberOfOpenFields] = fieldNameHashCode;
         openPartOffsets[this.numberOfOpenFields] = (openPartOffsets[numberOfOpenFields] << 32);
         openPartOffsets[numberOfOpenFields++] += openPartOutputStream.size();
-        openPartOutputStream.write(name.getBytes(), name.getStartIndex() + 1, name.getLength() - 1);
-        openPartOutputStream.write(value.getBytes(), value.getStartIndex(), value.getLength());
+        openPartOutputStream.write(name.getByteArray(), name.getStartOffset() + 1, name.getLength() - 1);
+        openPartOutputStream.write(value.getByteArray(), value.getStartOffset(), value.getLength());
     }
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/UnorderedListBuilder.java b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/UnorderedListBuilder.java
index 2f74f97..e8dbc29 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/builders/UnorderedListBuilder.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/builders/UnorderedListBuilder.java
@@ -10,7 +10,7 @@
 import edu.uci.ics.asterix.om.types.AUnorderedListType;
 import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
+import edu.uci.ics.hyracks.data.std.api.IValueReference;
 
 public class UnorderedListBuilder implements IAUnorderedListBuilder {
 
@@ -63,10 +63,10 @@
             this.offsets.add((short) outputStream.size());
         if (itemTypeTag == ATypeTag.ANY) {
             this.numberOfItems++;
-            this.outputStream.write(item.getBytes(), item.getStartIndex(), item.getLength());
-        } else if (item.getBytes()[0] != serNullTypeTag) {
+            this.outputStream.write(item.getByteArray(), item.getStartOffset(), item.getLength());
+        } else if (item.getByteArray()[0] != serNullTypeTag) {
             this.numberOfItems++;
-            this.outputStream.write(item.getBytes(), item.getStartIndex() + 1, item.getLength() - 1);
+            this.outputStream.write(item.getByteArray(), item.getStartOffset() + 1, item.getLength() - 1);
         }
     }
 
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/AFlatValueAccessor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/AFlatValueAccessor.java
deleted file mode 100644
index 83c3791..0000000
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/AFlatValueAccessor.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * 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;
-
-import edu.uci.ics.asterix.common.exceptions.AsterixException;
-import edu.uci.ics.asterix.om.types.IAType;
-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.container.IElementFactory;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
-
-public class AFlatValueAccessor extends AbstractBinaryAccessor {
-
-    public static IElementFactory<IBinaryAccessor, IAType> FACTORY = new IElementFactory<IBinaryAccessor, IAType>() {
-        public AFlatValueAccessor createElement(IAType type) {
-            return new AFlatValueAccessor();
-        }
-    };
-
-    private AFlatValueAccessor() {
-
-    }
-
-    @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();
-
-        byte[] data = getBytes();
-        int start = getStartIndex();
-        int len = getLength();
-        if ( len!= olen)
-            return false;
-        for (int i = 0; i < len; i++) {
-            if (data[start + i] != odata[ostart + i])
-                return false;
-        }
-        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/AbstractBinaryAccessor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/AbstractBinaryAccessor.java
deleted file mode 100644
index 3e027b8..0000000
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/AbstractBinaryAccessor.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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;
-
-import edu.uci.ics.asterix.runtime.accessors.base.IBinaryAccessor;
-
-public abstract class AbstractBinaryAccessor implements IBinaryAccessor {
-
-    private byte[] data;
-    private int start;
-    private int len;
-    
-    @Override
-    public byte[] getBytes() {
-        return data;
-    }
-
-    @Override
-    public int getLength() {
-        return len;
-    }
-
-    @Override
-    public int getStartIndex() {
-        return start;
-    }
-
-    @Override
-    public void reset(byte[] b, int start, int len) {
-        this.data = b;
-        this.start = start;
-        this.len = len;
-    }
-
-}
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
deleted file mode 100644
index 37d8b80..0000000
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/base/IBinaryAccessor.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * 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
deleted file mode 100644
index 54b0960..0000000
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/cast/ACastVisitor.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * 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.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.AbstractCollectionType;
-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.base.IBinaryAccessor;
-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<IBinaryAccessor, IAType, Boolean>> {
-
-    private Map<IBinaryAccessor, ARecordCaster> raccessorToCaster = new HashMap<IBinaryAccessor, ARecordCaster>();
-    private Map<IBinaryAccessor, AListCaster> laccessorToCaster = new HashMap<IBinaryAccessor, AListCaster>();
-
-    @Override
-    public Void visit(AListAccessor accessor, Triple<IBinaryAccessor, IAType, Boolean> arg) throws AsterixException {
-        AListCaster caster = laccessorToCaster.get(accessor);
-        if (caster == null) {
-            caster = new AListCaster();
-            laccessorToCaster.put(accessor, caster);
-        }
-        try {
-            caster.castList(accessor, arg.first, (AbstractCollectionType) arg.second, this);
-        } catch (Exception e) {
-            throw new AsterixException(e);
-        }
-        return null;
-    }
-
-    @Override
-    public Void visit(ARecordAccessor accessor, Triple<IBinaryAccessor, 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<IBinaryAccessor, IAType, Boolean> arg) {
-        // set the pointer for result
-        arg.first.reset(accessor.getBytes(), accessor.getStartIndex(), accessor.getLength());
-        return null;
-    }
-
-}
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
deleted file mode 100644
index e35421a..0000000
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/visitor/IBinaryAccessorVisitor.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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.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/aggregates/serializable/std/SerializableAvgAggregateDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableAvgAggregateDescriptor.java
index acc1e09..5c6aa8e 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableAvgAggregateDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableAvgAggregateDescriptor.java
@@ -91,35 +91,35 @@
                         if (inputVal.getLength() > 0) {
                             ++count;
                             ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
-                                    .deserialize(inputVal.getBytes()[0]);
+                                    .deserialize(inputVal.getByteArray()[0]);
                             switch (typeTag) {
                                 case INT8: {
-                                    byte val = AInt8SerializerDeserializer.getByte(inputVal.getBytes(), 1);
+                                    byte val = AInt8SerializerDeserializer.getByte(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT16: {
-                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getBytes(), 1);
+                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT32: {
-                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT64: {
-                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getBytes(), 1);
+                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case FLOAT: {
-                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getBytes(), 1);
+                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case DOUBLE: {
-                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getBytes(), 1);
+                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableGlobalAvgAggregateDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableGlobalAvgAggregateDescriptor.java
index b2c9be8..5b647d4 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableGlobalAvgAggregateDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableGlobalAvgAggregateDescriptor.java
@@ -118,7 +118,7 @@
 
                         inputVal.reset();
                         eval.evaluate(tuple);
-                        byte[] serBytes = inputVal.getBytes();
+                        byte[] serBytes = inputVal.getByteArray();
                         if (serBytes[0] == SER_NULL_TYPE_TAG)
                             metNull = true;
                         if (serBytes[0] != SER_RECORD_TYPE_TAG) {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableLocalAvgAggregateDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableLocalAvgAggregateDescriptor.java
index 9d65ad4..9b5e5e9 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableLocalAvgAggregateDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableLocalAvgAggregateDescriptor.java
@@ -121,35 +121,35 @@
                         if (inputVal.getLength() > 0) {
                             ++count;
                             ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
-                                    .deserialize(inputVal.getBytes()[0]);
+                                    .deserialize(inputVal.getByteArray()[0]);
                             switch (typeTag) {
                                 case INT8: {
-                                    byte val = AInt8SerializerDeserializer.getByte(inputVal.getBytes(), 1);
+                                    byte val = AInt8SerializerDeserializer.getByte(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT16: {
-                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getBytes(), 1);
+                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT32: {
-                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT64: {
-                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getBytes(), 1);
+                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case FLOAT: {
-                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getBytes(), 1);
+                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case DOUBLE: {
-                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getBytes(), 1);
+                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableSumAggregateDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableSumAggregateDescriptor.java
index ac1b4e1..4764cdb 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableSumAggregateDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/serializable/std/SerializableSumAggregateDescriptor.java
@@ -107,41 +107,41 @@
                         eval.evaluate(tuple);
                         if (inputVal.getLength() > 0) {
                             ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
-                                    .deserialize(inputVal.getBytes()[0]);
+                                    .deserialize(inputVal.getByteArray()[0]);
                             switch (typeTag) {
                                 case INT8: {
                                     metInt8s = true;
-                                    byte val = AInt8SerializerDeserializer.getByte(inputVal.getBytes(), 1);
+                                    byte val = AInt8SerializerDeserializer.getByte(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT16: {
                                     metInt16s = true;
-                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getBytes(), 1);
+                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT32: {
                                     metInt32s = true;
-                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT64: {
                                     metInt64s = true;
-                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getBytes(), 1);
+                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case FLOAT: {
                                     metFloats = true;
-                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getBytes(), 1);
+                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case DOUBLE: {
                                     metDoubles = true;
-                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getBytes(), 1);
+                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/AvgAggregateDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/AvgAggregateDescriptor.java
index 028a9a2..f5c9816 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/AvgAggregateDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/AvgAggregateDescriptor.java
@@ -120,35 +120,35 @@
                         if (inputVal.getLength() > 0) {
                             ++count;
                             ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
-                                    .deserialize(inputVal.getBytes()[0]);
+                                    .deserialize(inputVal.getByteArray()[0]);
                             switch (typeTag) {
                                 case INT8: {
-                                    byte val = AInt8SerializerDeserializer.getByte(inputVal.getBytes(), 1);
+                                    byte val = AInt8SerializerDeserializer.getByte(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT16: {
-                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getBytes(), 1);
+                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT32: {
-                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT64: {
-                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getBytes(), 1);
+                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case FLOAT: {
-                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getBytes(), 1);
+                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case DOUBLE: {
-                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getBytes(), 1);
+                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/GlobalAvgAggregateDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/GlobalAvgAggregateDescriptor.java
index f1d6fbd..530f790 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/GlobalAvgAggregateDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/GlobalAvgAggregateDescriptor.java
@@ -116,7 +116,7 @@
                     public void step(IFrameTupleReference tuple) throws AlgebricksException {
                         inputVal.reset();
                         eval.evaluate(tuple);
-                        byte[] serBytes = inputVal.getBytes();
+                        byte[] serBytes = inputVal.getByteArray();
                         if (serBytes[0] == SER_NULL_TYPE_TAG)
                             metNull = true;
                         if (serBytes[0] != SER_RECORD_TYPE_TAG) {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/LocalAvgAggregateDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/LocalAvgAggregateDescriptor.java
index 88d917f..5523b0d 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/LocalAvgAggregateDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/LocalAvgAggregateDescriptor.java
@@ -121,35 +121,35 @@
                         if (inputVal.getLength() > 0) {
                             ++count;
                             ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
-                                    .deserialize(inputVal.getBytes()[0]);
+                                    .deserialize(inputVal.getByteArray()[0]);
                             switch (typeTag) {
                                 case INT8: {
-                                    byte val = AInt8SerializerDeserializer.getByte(inputVal.getBytes(), 1);
+                                    byte val = AInt8SerializerDeserializer.getByte(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT16: {
-                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getBytes(), 1);
+                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT32: {
-                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT64: {
-                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getBytes(), 1);
+                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case FLOAT: {
-                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getBytes(), 1);
+                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case DOUBLE: {
-                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getBytes(), 1);
+                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/MaxAggregateDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/MaxAggregateDescriptor.java
index 6f7ac58..43a8cb1 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/MaxAggregateDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/MaxAggregateDescriptor.java
@@ -106,7 +106,7 @@
                         eval.evaluate(tuple);
                         if (inputVal.getLength() > 0) {
                             ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
-                                    .deserialize(inputVal.getBytes()[0]);
+                                    .deserialize(inputVal.getByteArray()[0]);
                             switch (typeTag) {
                                 case INT8: {
                                     metInt8s = true;
@@ -114,35 +114,35 @@
                                 }
                                 case INT16: {
                                     metInt16s = true;
-                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getBytes(), 1);
+                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getByteArray(), 1);
                                     if (val > shortVal)
                                         shortVal = val;
                                     throw new NotImplementedException("no implementation for int16's comparator");
                                 }
                                 case INT32: {
                                     metInt32s = true;
-                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getByteArray(), 1);
                                     if (val > intVal)
                                         intVal = val;
                                     break;
                                 }
                                 case INT64: {
                                     metInt64s = true;
-                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getBytes(), 1);
+                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getByteArray(), 1);
                                     if (val > longVal)
                                         longVal = val;
                                     break;
                                 }
                                 case FLOAT: {
                                     metFloats = true;
-                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getBytes(), 1);
+                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getByteArray(), 1);
                                     if (val > floatVal)
                                         floatVal = val;
                                     break;
                                 }
                                 case DOUBLE: {
                                     metDoubles = true;
-                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getBytes(), 1);
+                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getByteArray(), 1);
                                     if (val > doubleVal)
                                         doubleVal = val;
                                     break;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/MinAggregateDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/MinAggregateDescriptor.java
index 2b36918..8654076 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/MinAggregateDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/MinAggregateDescriptor.java
@@ -106,7 +106,7 @@
                         eval.evaluate(tuple);
                         if (inputVal.getLength() > 0) {
                             ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
-                                    .deserialize(inputVal.getBytes()[0]);
+                                    .deserialize(inputVal.getByteArray()[0]);
                             switch (typeTag) {
                                 case INT8: {
                                     metInt8s = true;
@@ -114,35 +114,35 @@
                                 }
                                 case INT16: {
                                     metInt16s = true;
-                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getBytes(), 1);
+                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getByteArray(), 1);
                                     if (val < shortVal)
                                         shortVal = val;
                                     throw new NotImplementedException("no implementation for int16's comparator");
                                 }
                                 case INT32: {
                                     metInt32s = true;
-                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getByteArray(), 1);
                                     if (val < intVal)
                                         intVal = val;
                                     break;
                                 }
                                 case INT64: {
                                     metInt64s = true;
-                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getBytes(), 1);
+                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getByteArray(), 1);
                                     if (val < longVal)
                                         longVal = val;
                                     break;
                                 }
                                 case FLOAT: {
                                     metFloats = true;
-                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getBytes(), 1);
+                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getByteArray(), 1);
                                     if (val < floatVal)
                                         floatVal = val;
                                     break;
                                 }
                                 case DOUBLE: {
                                     metDoubles = true;
-                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getBytes(), 1);
+                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getByteArray(), 1);
                                     if (val < doubleVal)
                                         doubleVal = val;
                                     break;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/SumAggregateDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/SumAggregateDescriptor.java
index e51badd..129d0b7 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/SumAggregateDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/aggregates/std/SumAggregateDescriptor.java
@@ -98,41 +98,41 @@
                         eval.evaluate(tuple);
                         if (inputVal.getLength() > 0) {
                             ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
-                                    .deserialize(inputVal.getBytes()[0]);
+                                    .deserialize(inputVal.getByteArray()[0]);
                             switch (typeTag) {
                                 case INT8: {
                                     metInt8s = true;
-                                    byte val = AInt8SerializerDeserializer.getByte(inputVal.getBytes(), 1);
+                                    byte val = AInt8SerializerDeserializer.getByte(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT16: {
                                     metInt16s = true;
-                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getBytes(), 1);
+                                    short val = AInt16SerializerDeserializer.getShort(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT32: {
                                     metInt32s = true;
-                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                                    int val = AInt32SerializerDeserializer.getInt(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case INT64: {
                                     metInt64s = true;
-                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getBytes(), 1);
+                                    long val = AInt64SerializerDeserializer.getLong(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case FLOAT: {
                                     metFloats = true;
-                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getBytes(), 1);
+                                    float val = AFloatSerializerDeserializer.getFloat(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
                                 case DOUBLE: {
                                     metDoubles = true;
-                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getBytes(), 1);
+                                    double val = ADoubleSerializerDeserializer.getDouble(inputVal.getByteArray(), 1);
                                     sum += val;
                                     break;
                                 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/base/AsterixTupleFilter.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/base/AsterixTupleFilter.java
index 8bb0b9f..3ef8cf3 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/base/AsterixTupleFilter.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/base/AsterixTupleFilter.java
@@ -39,6 +39,6 @@
 	public boolean accept(IFrameTupleReference tuple) throws Exception {
 		evalOut.reset();
 		eval.evaluate(tuple);
-		return boolInspector.getBooleanValue(evalOut.getBytes(), 0, 2);
+		return boolInspector.getBooleanValue(evalOut.getByteArray(), 0, 2);
 	}
 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/ClosedRecordConstructorEvalFactory.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/ClosedRecordConstructorEvalFactory.java
index 15ab112..2c14bd0 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/ClosedRecordConstructorEvalFactory.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/ClosedRecordConstructorEvalFactory.java
@@ -67,7 +67,7 @@
                 for (int i = 0; i < evalFields.length; i++) {
                     fieldValueBuffer.reset();
                     evalFields[i].evaluate(tuple);
-                    if (fieldValueBuffer.getBytes()[0] != SER_NULL_TYPE_TAG) {
+                    if (fieldValueBuffer.getByteArray()[0] != SER_NULL_TYPE_TAG) {
                         recBuilder.addField(i, fieldValueBuffer);
                     }
                 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/CreateMBREvalFactory.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/CreateMBREvalFactory.java
index d7404bb..165cd56 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/CreateMBREvalFactory.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/CreateMBREvalFactory.java
@@ -63,11 +63,11 @@
 
                 try {
 
-                    int dimension = AInt32SerializerDeserializer.getInt(outInput1.getBytes(), 1);
-                    int coordinate = AInt32SerializerDeserializer.getInt(outInput2.getBytes(), 1);
+                    int dimension = AInt32SerializerDeserializer.getInt(outInput1.getByteArray(), 1);
+                    int coordinate = AInt32SerializerDeserializer.getInt(outInput2.getByteArray(), 1);
                     double value;
                     if (dimension == 2) {
-                        ATypeTag tag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(outInput0.getBytes()[0]);
+                        ATypeTag tag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(outInput0.getByteArray()[0]);
                         switch (tag) {
                             case POINT:
                                 switch (coordinate) {
@@ -75,7 +75,7 @@
                                             // for
                                             // max x, and 3 for max y
                                     case 2: {
-                                        double x = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double x = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 APointSerializerDeserializer.getCoordinateOffset(Coordinate.X));
 
                                         value = x;
@@ -83,7 +83,7 @@
                                         break;
                                     case 1:
                                     case 3: {
-                                        double y = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double y = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 APointSerializerDeserializer.getCoordinateOffset(Coordinate.Y));
 
                                         value = y;
@@ -101,9 +101,9 @@
                                     case 0: {
                                         value = Double.MAX_VALUE;
                                         double startX = ADoubleSerializerDeserializer
-                                                .getDouble(outInput0.getBytes(), ALineSerializerDeserializer
+                                                .getDouble(outInput0.getByteArray(), ALineSerializerDeserializer
                                                         .getStartPointCoordinateOffset(Coordinate.X));
-                                        double endX = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double endX = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ALineSerializerDeserializer.getEndPointCoordinateOffset(Coordinate.X));
 
                                         value = Math.min(Math.min(startX, endX), value);
@@ -112,9 +112,9 @@
                                     case 1: {
                                         value = Double.MAX_VALUE;
                                         double startY = ADoubleSerializerDeserializer
-                                                .getDouble(outInput0.getBytes(), ALineSerializerDeserializer
+                                                .getDouble(outInput0.getByteArray(), ALineSerializerDeserializer
                                                         .getStartPointCoordinateOffset(Coordinate.Y));
-                                        double endY = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double endY = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ALineSerializerDeserializer.getEndPointCoordinateOffset(Coordinate.Y));
 
                                         value = Math.min(Math.min(startY, endY), value);
@@ -123,9 +123,9 @@
                                     case 2: {
                                         value = Double.MIN_VALUE;
                                         double startX = ADoubleSerializerDeserializer
-                                                .getDouble(outInput0.getBytes(), ALineSerializerDeserializer
+                                                .getDouble(outInput0.getByteArray(), ALineSerializerDeserializer
                                                         .getStartPointCoordinateOffset(Coordinate.X));
-                                        double endX = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double endX = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ALineSerializerDeserializer.getEndPointCoordinateOffset(Coordinate.X));
 
                                         value = Math.max(Math.min(startX, endX), value);
@@ -134,9 +134,9 @@
                                     case 3: {
                                         value = Double.MIN_VALUE;
                                         double startY = ADoubleSerializerDeserializer
-                                                .getDouble(outInput0.getBytes(), ALineSerializerDeserializer
+                                                .getDouble(outInput0.getByteArray(), ALineSerializerDeserializer
                                                         .getStartPointCoordinateOffset(Coordinate.Y));
-                                        double endY = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double endY = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ALineSerializerDeserializer.getEndPointCoordinateOffset(Coordinate.Y));
 
                                         value = Math.max(Math.min(startY, endY), value);
@@ -149,14 +149,14 @@
                                 }
                                 break;
                             case POLYGON:
-                                int numOfPoints = AInt16SerializerDeserializer.getShort(outInput0.getBytes(),
+                                int numOfPoints = AInt16SerializerDeserializer.getShort(outInput0.getByteArray(),
                                         APolygonSerializerDeserializer.getNumberOfPointsOffset());
                                 switch (coordinate) {
                                     case 0: {
                                         value = Double.MAX_VALUE;
                                         for (int i = 0; i < numOfPoints; i++) {
                                             double x = ADoubleSerializerDeserializer
-                                                    .getDouble(outInput0.getBytes(), APolygonSerializerDeserializer
+                                                    .getDouble(outInput0.getByteArray(), APolygonSerializerDeserializer
                                                             .getCoordinateOffset(i, Coordinate.X));
                                             value = Math.min(x, value);
                                         }
@@ -166,7 +166,7 @@
                                         value = Double.MAX_VALUE;
                                         for (int i = 0; i < numOfPoints; i++) {
                                             double y = ADoubleSerializerDeserializer
-                                                    .getDouble(outInput0.getBytes(), APolygonSerializerDeserializer
+                                                    .getDouble(outInput0.getByteArray(), APolygonSerializerDeserializer
                                                             .getCoordinateOffset(i, Coordinate.Y));
                                             value = Math.min(y, value);
                                         }
@@ -176,7 +176,7 @@
                                         value = Double.MIN_VALUE;
                                         for (int i = 0; i < numOfPoints; i++) {
                                             double x = ADoubleSerializerDeserializer
-                                                    .getDouble(outInput0.getBytes(), APolygonSerializerDeserializer
+                                                    .getDouble(outInput0.getByteArray(), APolygonSerializerDeserializer
                                                             .getCoordinateOffset(i, Coordinate.X));
                                             value = Math.max(x, value);
                                         }
@@ -186,7 +186,7 @@
                                         value = Double.MIN_VALUE;
                                         for (int i = 0; i < numOfPoints; i++) {
                                             double y = ADoubleSerializerDeserializer
-                                                    .getDouble(outInput0.getBytes(), APolygonSerializerDeserializer
+                                                    .getDouble(outInput0.getByteArray(), APolygonSerializerDeserializer
                                                             .getCoordinateOffset(i, Coordinate.Y));
                                             value = Math.max(y, value);
                                         }
@@ -201,10 +201,10 @@
                             case CIRCLE:
                                 switch (coordinate) {
                                     case 0: {
-                                        double x = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double x = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ACircleSerializerDeserializer
                                                         .getCenterPointCoordinateOffset(Coordinate.X));
-                                        double radius = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double radius = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ACircleSerializerDeserializer
                                                         .getCenterPointCoordinateOffset(Coordinate.X));
 
@@ -212,10 +212,10 @@
                                     }
                                         break;
                                     case 1: {
-                                        double y = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double y = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ACircleSerializerDeserializer
                                                         .getCenterPointCoordinateOffset(Coordinate.Y));
-                                        double radius = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double radius = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ACircleSerializerDeserializer
                                                         .getCenterPointCoordinateOffset(Coordinate.Y));
 
@@ -223,10 +223,10 @@
                                     }
                                         break;
                                     case 2: {
-                                        double x = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double x = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ACircleSerializerDeserializer
                                                         .getCenterPointCoordinateOffset(Coordinate.X));
-                                        double radius = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double radius = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ACircleSerializerDeserializer
                                                         .getCenterPointCoordinateOffset(Coordinate.X));
 
@@ -234,10 +234,10 @@
                                     }
                                         break;
                                     case 3: {
-                                        double y = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double y = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ACircleSerializerDeserializer
                                                         .getCenterPointCoordinateOffset(Coordinate.Y));
-                                        double radius = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        double radius = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ACircleSerializerDeserializer
                                                         .getCenterPointCoordinateOffset(Coordinate.Y));
 
@@ -254,25 +254,25 @@
                                 value = Double.MAX_VALUE;
                                 switch (coordinate) {
                                     case 0: {
-                                        value = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        value = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ARectangleSerializerDeserializer
                                                         .getBottomLeftCoordinateOffset(Coordinate.X));
                                     }
                                         break;
                                     case 1: {
-                                        value = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        value = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ARectangleSerializerDeserializer
                                                         .getBottomLeftCoordinateOffset(Coordinate.Y));
                                     }
                                         break;
                                     case 2: {
-                                        value = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        value = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ARectangleSerializerDeserializer
                                                         .getUpperRightCoordinateOffset(Coordinate.X));
                                     }
                                         break;
                                     case 3: {
-                                        value = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                        value = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                                 ARectangleSerializerDeserializer
                                                         .getUpperRightCoordinateOffset(Coordinate.Y));
                                     }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/EditDistanceEvaluator.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/EditDistanceEvaluator.java
index 0f8be91..57331db 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/EditDistanceEvaluator.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/EditDistanceEvaluator.java
@@ -56,15 +56,15 @@
         if (!checkArgTypes(firstTypeTag, secondTypeTag))
             return;
 
-        itemTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getBytes()[firstStart + 1]);
+        itemTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getByteArray()[firstStart + 1]);
         if (itemTypeTag == ATypeTag.ANY)
             throw new AlgebricksException("\n Edit Distance can only be called on homogenous lists");
 
-        itemTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getBytes()[secondStart + 1]);
+        itemTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getByteArray()[secondStart + 1]);
         if (itemTypeTag == ATypeTag.ANY)
             throw new AlgebricksException("\n Edit Distance can only be called on homogenous lists");
 
-        editDistance = computeResult(argOut.getBytes(), firstStart, secondStart, firstTypeTag);
+        editDistance = computeResult(argOut.getByteArray(), firstStart, secondStart, firstTypeTag);
 
         try {
             writeResult(editDistance);
@@ -81,8 +81,8 @@
         secondStart = argOut.getLength();
         secondStringEval.evaluate(tuple);
 
-        firstTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getBytes()[firstStart]);
-        secondTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getBytes()[secondStart]);
+        firstTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getByteArray()[firstStart]);
+        secondTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getByteArray()[secondStart]);
     }
 
     protected int computeResult(byte[] bytes, int firstStart, int secondStart, ATypeTag argType)
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/FieldAccessByIndexEvalFactory.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/FieldAccessByIndexEvalFactory.java
index 06bcedf..49a1d99 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/FieldAccessByIndexEvalFactory.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/FieldAccessByIndexEvalFactory.java
@@ -73,7 +73,7 @@
                     eval0.evaluate(tuple);
                     outInput1.reset();
                     eval1.evaluate(tuple);
-                    byte[] serRecord = outInput0.getBytes();
+                    byte[] serRecord = outInput0.getByteArray();
 
                     if (serRecord[0] == SER_NULL_TYPE_TAG) {
                         nullSerde.serialize(ANull.NULL, out);
@@ -85,7 +85,7 @@
                                 + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(serRecord[0]));
                     }
 
-                    fieldIndex = IntegerSerializerDeserializer.getInt(outInput1.getBytes(), 1);
+                    fieldIndex = IntegerSerializerDeserializer.getInt(outInput1.getByteArray(), 1);
                     fieldValueOffset = ARecordSerializerDeserializer.getFieldOffsetById(serRecord, fieldIndex,
                             nullBitmapSize, recordType.isOpen());
 
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/GramTokensEvaluator.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/GramTokensEvaluator.java
index 4713c38..4b392bd 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/GramTokensEvaluator.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/GramTokensEvaluator.java
@@ -59,7 +59,7 @@
         int prePostOff = argOut.getLength();
         prePostEval.evaluate(tuple);
 
-        byte[] bytes = argOut.getBytes();
+        byte[] bytes = argOut.getByteArray();
         int gramLength = IntegerSerializerDeserializer.getInt(bytes, gramLengthOff + typeIndicatorSize);
         tokenizer.setGramlength(gramLength);
         boolean prePost = BooleanSerializerDeserializer.getBoolean(bytes, prePostOff + typeIndicatorSize);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/SimilarityJaccardEvaluator.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/SimilarityJaccardEvaluator.java
index 932036e..0c0c36d 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/SimilarityJaccardEvaluator.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/SimilarityJaccardEvaluator.java
@@ -63,7 +63,7 @@
         runArgEvals(tuple);
         if (!checkArgTypes(firstTypeTag, secondTypeTag))
             return;
-        jaccSim = computeResult(argOut.getBytes(), firstStart, secondStart, firstTypeTag);
+        jaccSim = computeResult(argOut.getByteArray(), firstStart, secondStart, firstTypeTag);
         try {
             writeResult(jaccSim);
         } catch (IOException e) {
@@ -79,8 +79,8 @@
         secondStart = argOut.getLength();
         secondOrdListEval.evaluate(tuple);
 
-        firstTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getBytes()[firstStart]);
-        secondTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getBytes()[secondStart]);
+        firstTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getByteArray()[firstStart]);
+        secondTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getByteArray()[secondStart]);
     }
 
     protected float computeResult(byte[] bytes, int firstStart, int secondStart, ATypeTag argType)
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/SimilarityJaccardPrefixEvaluator.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/SimilarityJaccardPrefixEvaluator.java
index 4884806..8961257 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/SimilarityJaccardPrefixEvaluator.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/SimilarityJaccardPrefixEvaluator.java
@@ -74,7 +74,7 @@
         sim = 0;
         inputVal.reset();
         evalThreshold.evaluate(tuple);
-        float similarityThreshold = AFloatSerializerDeserializer.getFloat(inputVal.getBytes(), 1);
+        float similarityThreshold = AFloatSerializerDeserializer.getFloat(inputVal.getByteArray(), 1);
 
         if (similarityThreshold != similarityThresholdCache || similarityFilters == null) {
             similarityFilters = new SimilarityFiltersJaccard(similarityThreshold);
@@ -83,11 +83,11 @@
 
         inputVal.reset();
         evalLen1.evaluate(tuple);
-        int length1 = IntegerSerializerDeserializer.getInt(inputVal.getBytes(), 1);
+        int length1 = IntegerSerializerDeserializer.getInt(inputVal.getByteArray(), 1);
 
         inputVal.reset();
         evalLen2.evaluate(tuple);
-        int length2 = IntegerSerializerDeserializer.getInt(inputVal.getBytes(), 1);
+        int length2 = IntegerSerializerDeserializer.getInt(inputVal.getByteArray(), 1);
 
         //
         // -- - length filter - --
@@ -100,7 +100,7 @@
             inputVal.reset();
             evalTokens1.evaluate(tuple);
 
-            byte[] serList = inputVal.getBytes();
+            byte[] serList = inputVal.getByteArray();
             if (serList[0] != SER_ORDEREDLIST_TYPE_TAG && serList[0] != SER_UNORDEREDLIST_TYPE_TAG) {
                 throw new AlgebricksException("Scan collection is not defined for values of type"
                         + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(serList[0]));
@@ -108,7 +108,7 @@
 
             int lengthTokens1;
             if (serList[0] == SER_ORDEREDLIST_TYPE_TAG) {
-                lengthTokens1 = AOrderedListSerializerDeserializer.getNumberOfItems(inputVal.getBytes());
+                lengthTokens1 = AOrderedListSerializerDeserializer.getNumberOfItems(inputVal.getByteArray());
                 // read tokens
                 for (i = 0; i < lengthTokens1; i++) {
                     int itemOffset;
@@ -120,7 +120,7 @@
                     tokens1.add(IntegerSerializerDeserializer.getInt(serList, itemOffset));
                 }
             } else {
-                lengthTokens1 = AUnorderedListSerializerDeserializer.getNumberOfItems(inputVal.getBytes());
+                lengthTokens1 = AUnorderedListSerializerDeserializer.getNumberOfItems(inputVal.getByteArray());
                 // read tokens
                 for (i = 0; i < lengthTokens1; i++) {
                     int itemOffset;
@@ -142,7 +142,7 @@
             inputVal.reset();
             evalTokens2.evaluate(tuple);
 
-            serList = inputVal.getBytes();
+            serList = inputVal.getByteArray();
             if (serList[0] != SER_ORDEREDLIST_TYPE_TAG && serList[0] != SER_UNORDEREDLIST_TYPE_TAG) {
                 throw new AlgebricksException("Scan collection is not defined for values of type"
                         + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(serList[0]));
@@ -150,7 +150,7 @@
 
             int lengthTokens2;
             if (serList[0] == SER_ORDEREDLIST_TYPE_TAG) {
-                lengthTokens2 = AOrderedListSerializerDeserializer.getNumberOfItems(inputVal.getBytes());
+                lengthTokens2 = AOrderedListSerializerDeserializer.getNumberOfItems(inputVal.getByteArray());
                 // read tokens
                 for (i = 0; i < lengthTokens2; i++) {
                     int itemOffset;
@@ -162,7 +162,7 @@
                     tokens2.add(IntegerSerializerDeserializer.getInt(serList, itemOffset));
                 }
             } else {
-                lengthTokens2 = AUnorderedListSerializerDeserializer.getNumberOfItems(inputVal.getBytes());
+                lengthTokens2 = AUnorderedListSerializerDeserializer.getNumberOfItems(inputVal.getByteArray());
                 // read tokens
                 for (i = 0; i < lengthTokens2; i++) {
                     int itemOffset;
@@ -182,7 +182,7 @@
             // -- - token prefix - --
             inputVal.reset();
             evalTokenPrefix.evaluate(tuple);
-            int tokenPrefix = IntegerSerializerDeserializer.getInt(inputVal.getBytes(), 1);
+            int tokenPrefix = IntegerSerializerDeserializer.getInt(inputVal.getByteArray(), 1);
 
             //
             // -- - position filter - --
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/WordTokensEvaluator.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/WordTokensEvaluator.java
index 800c4ee..d9fde50 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/WordTokensEvaluator.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/common/WordTokensEvaluator.java
@@ -43,7 +43,7 @@
     public void evaluate(IFrameTupleReference tuple) throws AlgebricksException {
         argOut.reset();
         stringEval.evaluate(tuple);
-        byte[] bytes = argOut.getBytes();
+        byte[] bytes = argOut.getByteArray();
         tokenizer.reset(bytes, 0, argOut.getLength());
         tokenBuffer.reset();
 
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/comparisons/AbstractComparisonEvaluator.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/comparisons/AbstractComparisonEvaluator.java
index 958b04b..5082c67 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/comparisons/AbstractComparisonEvaluator.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/comparisons/AbstractComparisonEvaluator.java
@@ -74,7 +74,7 @@
         if (outLeft.getLength() == 0) {
             isLeftNull = true;
         } else {
-            typeTag1 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(outLeft.getBytes()[0]);
+            typeTag1 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(outLeft.getByteArray()[0]);
             if (typeTag1 == ATypeTag.NULL) {
                 isLeftNull = true;
             }
@@ -82,7 +82,7 @@
         if (outRight.getLength() == 0) {
             isRightNull = true;
         } else {
-            typeTag2 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(outRight.getBytes()[0]);
+            typeTag2 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(outRight.getByteArray()[0]);
             if (typeTag2 == ATypeTag.NULL) {
                 isRightNull = true;
             }
@@ -130,8 +130,8 @@
         if (typeTag2 == ATypeTag.NULL) {
             return ComparisonResult.GREATER_THAN;
         } else if (typeTag2 == ATypeTag.DATETIME) {
-            int result = dateTimeBinaryComp.compare(outLeft.getBytes(), 1, outLeft.getLength() - 1,
-                    outRight.getBytes(), 1, outRight.getLength() - 1);
+            int result = dateTimeBinaryComp.compare(outLeft.getByteArray(), 1, outLeft.getLength() - 1,
+                    outRight.getByteArray(), 1, outRight.getLength() - 1);
             if (result == 0)
                 return ComparisonResult.EQUAL;
             else if (result < 0)
@@ -144,8 +144,8 @@
 
     private ComparisonResult compareBooleanWithArg(ATypeTag typeTag2) throws AlgebricksException {
         if (typeTag2 == ATypeTag.BOOLEAN) {
-            byte b0 = outLeft.getBytes()[1];
-            byte b1 = outRight.getBytes()[1];
+            byte b0 = outLeft.getByteArray()[1];
+            byte b1 = outRight.getByteArray()[1];
             return compareByte(b0, b1);
         }
         throw new AlgebricksException("Comparison is undefined between types ABoolean and " + typeTag2 + " .");
@@ -153,7 +153,7 @@
 
     private ComparisonResult compareStringWithArg(ATypeTag typeTag2) throws AlgebricksException {
         if (typeTag2 == ATypeTag.STRING) {
-            int result = strBinaryComp.compare(outLeft.getBytes(), 1, outLeft.getLength() - 1, outRight.getBytes(), 1,
+            int result = strBinaryComp.compare(outLeft.getByteArray(), 1, outLeft.getLength() - 1, outRight.getByteArray(), 1,
                     outRight.getLength() - 1);
             if (result == 0)
                 return ComparisonResult.EQUAL;
@@ -166,30 +166,30 @@
     }
 
     private ComparisonResult compareDoubleWithArg(ATypeTag typeTag2) throws AlgebricksException {
-        double s = ADoubleSerializerDeserializer.getDouble(outLeft.getBytes(), 1);
+        double s = ADoubleSerializerDeserializer.getDouble(outLeft.getByteArray(), 1);
         switch (typeTag2) {
             case INT8: {
-                byte v2 = AInt8SerializerDeserializer.getByte(outRight.getBytes(), 1);
+                byte v2 = AInt8SerializerDeserializer.getByte(outRight.getByteArray(), 1);
                 return compareDouble(s, v2);
             }
             case INT16: {
-                short v2 = AInt16SerializerDeserializer.getShort(outRight.getBytes(), 1);
+                short v2 = AInt16SerializerDeserializer.getShort(outRight.getByteArray(), 1);
                 return compareDouble(s, v2);
             }
             case INT32: {
-                int v2 = AInt32SerializerDeserializer.getInt(outRight.getBytes(), 1);
+                int v2 = AInt32SerializerDeserializer.getInt(outRight.getByteArray(), 1);
                 return compareDouble(s, v2);
             }
             case INT64: {
-                long v2 = AInt64SerializerDeserializer.getLong(outRight.getBytes(), 1);
+                long v2 = AInt64SerializerDeserializer.getLong(outRight.getByteArray(), 1);
                 return compareDouble(s, v2);
             }
             case FLOAT: {
-                float v2 = AFloatSerializerDeserializer.getFloat(outRight.getBytes(), 1);
+                float v2 = AFloatSerializerDeserializer.getFloat(outRight.getByteArray(), 1);
                 return compareDouble(s, v2);
             }
             case DOUBLE: {
-                double v2 = ADoubleSerializerDeserializer.getDouble(outRight.getBytes(), 1);
+                double v2 = ADoubleSerializerDeserializer.getDouble(outRight.getByteArray(), 1);
                 return compareDouble(s, v2);
             }
             default: {
@@ -199,30 +199,30 @@
     }
 
     private ComparisonResult compareFloatWithArg(ATypeTag typeTag2) throws AlgebricksException {
-        float s = FloatSerializerDeserializer.getFloat(outLeft.getBytes(), 1);
+        float s = FloatSerializerDeserializer.getFloat(outLeft.getByteArray(), 1);
         switch (typeTag2) {
             case INT8: {
-                byte v2 = AInt8SerializerDeserializer.getByte(outRight.getBytes(), 1);
+                byte v2 = AInt8SerializerDeserializer.getByte(outRight.getByteArray(), 1);
                 return compareFloat(s, v2);
             }
             case INT16: {
-                short v2 = AInt16SerializerDeserializer.getShort(outRight.getBytes(), 1);
+                short v2 = AInt16SerializerDeserializer.getShort(outRight.getByteArray(), 1);
                 return compareFloat(s, v2);
             }
             case INT32: {
-                int v2 = AInt32SerializerDeserializer.getInt(outRight.getBytes(), 1);
+                int v2 = AInt32SerializerDeserializer.getInt(outRight.getByteArray(), 1);
                 return compareFloat(s, v2);
             }
             case INT64: {
-                long v2 = AInt64SerializerDeserializer.getLong(outRight.getBytes(), 1);
+                long v2 = AInt64SerializerDeserializer.getLong(outRight.getByteArray(), 1);
                 return compareFloat(s, v2);
             }
             case FLOAT: {
-                float v2 = AFloatSerializerDeserializer.getFloat(outRight.getBytes(), 1);
+                float v2 = AFloatSerializerDeserializer.getFloat(outRight.getByteArray(), 1);
                 return compareFloat(s, v2);
             }
             case DOUBLE: {
-                double v2 = ADoubleSerializerDeserializer.getDouble(outRight.getBytes(), 1);
+                double v2 = ADoubleSerializerDeserializer.getDouble(outRight.getByteArray(), 1);
                 return compareDouble(s, v2);
             }
             default: {
@@ -232,30 +232,30 @@
     }
 
     private ComparisonResult compareInt64WithArg(ATypeTag typeTag2) throws AlgebricksException {
-        long s = AInt64SerializerDeserializer.getLong(outLeft.getBytes(), 1);
+        long s = AInt64SerializerDeserializer.getLong(outLeft.getByteArray(), 1);
         switch (typeTag2) {
             case INT8: {
-                byte v2 = AInt8SerializerDeserializer.getByte(outRight.getBytes(), 1);
+                byte v2 = AInt8SerializerDeserializer.getByte(outRight.getByteArray(), 1);
                 return compareLong(s, v2);
             }
             case INT16: {
-                short v2 = AInt16SerializerDeserializer.getShort(outRight.getBytes(), 1);
+                short v2 = AInt16SerializerDeserializer.getShort(outRight.getByteArray(), 1);
                 return compareLong(s, v2);
             }
             case INT32: {
-                int v2 = AInt32SerializerDeserializer.getInt(outRight.getBytes(), 1);
+                int v2 = AInt32SerializerDeserializer.getInt(outRight.getByteArray(), 1);
                 return compareLong(s, v2);
             }
             case INT64: {
-                long v2 = AInt64SerializerDeserializer.getLong(outRight.getBytes(), 1);
+                long v2 = AInt64SerializerDeserializer.getLong(outRight.getByteArray(), 1);
                 return compareLong(s, v2);
             }
             case FLOAT: {
-                float v2 = AFloatSerializerDeserializer.getFloat(outRight.getBytes(), 1);
+                float v2 = AFloatSerializerDeserializer.getFloat(outRight.getByteArray(), 1);
                 return compareFloat(s, v2);
             }
             case DOUBLE: {
-                double v2 = ADoubleSerializerDeserializer.getDouble(outRight.getBytes(), 1);
+                double v2 = ADoubleSerializerDeserializer.getDouble(outRight.getByteArray(), 1);
                 return compareDouble(s, v2);
             }
             default: {
@@ -265,30 +265,30 @@
     }
 
     private ComparisonResult compareInt32WithArg(ATypeTag typeTag2) throws AlgebricksException {
-        int s = IntegerSerializerDeserializer.getInt(outLeft.getBytes(), 1);
+        int s = IntegerSerializerDeserializer.getInt(outLeft.getByteArray(), 1);
         switch (typeTag2) {
             case INT8: {
-                byte v2 = AInt8SerializerDeserializer.getByte(outRight.getBytes(), 1);
+                byte v2 = AInt8SerializerDeserializer.getByte(outRight.getByteArray(), 1);
                 return compareInt(s, v2);
             }
             case INT16: {
-                short v2 = AInt16SerializerDeserializer.getShort(outRight.getBytes(), 1);
+                short v2 = AInt16SerializerDeserializer.getShort(outRight.getByteArray(), 1);
                 return compareInt(s, v2);
             }
             case INT32: {
-                int v2 = AInt32SerializerDeserializer.getInt(outRight.getBytes(), 1);
+                int v2 = AInt32SerializerDeserializer.getInt(outRight.getByteArray(), 1);
                 return compareInt(s, v2);
             }
             case INT64: {
-                long v2 = AInt64SerializerDeserializer.getLong(outRight.getBytes(), 1);
+                long v2 = AInt64SerializerDeserializer.getLong(outRight.getByteArray(), 1);
                 return compareLong(s, v2);
             }
             case FLOAT: {
-                float v2 = AFloatSerializerDeserializer.getFloat(outRight.getBytes(), 1);
+                float v2 = AFloatSerializerDeserializer.getFloat(outRight.getByteArray(), 1);
                 return compareFloat(s, v2);
             }
             case DOUBLE: {
-                double v2 = ADoubleSerializerDeserializer.getDouble(outRight.getBytes(), 1);
+                double v2 = ADoubleSerializerDeserializer.getDouble(outRight.getByteArray(), 1);
                 return compareDouble(s, v2);
             }
             default: {
@@ -298,30 +298,30 @@
     }
 
     private ComparisonResult compareInt16WithArg(ATypeTag typeTag2) throws AlgebricksException {
-        short s = AInt16SerializerDeserializer.getShort(outLeft.getBytes(), 1);
+        short s = AInt16SerializerDeserializer.getShort(outLeft.getByteArray(), 1);
         switch (typeTag2) {
             case INT8: {
-                byte v2 = AInt8SerializerDeserializer.getByte(outRight.getBytes(), 1);
+                byte v2 = AInt8SerializerDeserializer.getByte(outRight.getByteArray(), 1);
                 return compareShort(s, v2);
             }
             case INT16: {
-                short v2 = AInt16SerializerDeserializer.getShort(outRight.getBytes(), 1);
+                short v2 = AInt16SerializerDeserializer.getShort(outRight.getByteArray(), 1);
                 return compareShort(s, v2);
             }
             case INT32: {
-                int v2 = AInt32SerializerDeserializer.getInt(outRight.getBytes(), 1);
+                int v2 = AInt32SerializerDeserializer.getInt(outRight.getByteArray(), 1);
                 return compareInt(s, v2);
             }
             case INT64: {
-                long v2 = AInt64SerializerDeserializer.getLong(outRight.getBytes(), 1);
+                long v2 = AInt64SerializerDeserializer.getLong(outRight.getByteArray(), 1);
                 return compareLong(s, v2);
             }
             case FLOAT: {
-                float v2 = AFloatSerializerDeserializer.getFloat(outRight.getBytes(), 1);
+                float v2 = AFloatSerializerDeserializer.getFloat(outRight.getByteArray(), 1);
                 return compareFloat(s, v2);
             }
             case DOUBLE: {
-                double v2 = ADoubleSerializerDeserializer.getDouble(outRight.getBytes(), 1);
+                double v2 = ADoubleSerializerDeserializer.getDouble(outRight.getByteArray(), 1);
                 return compareDouble(s, v2);
             }
             default: {
@@ -331,30 +331,30 @@
     }
 
     private ComparisonResult compareInt8WithArg(ATypeTag typeTag2) throws AlgebricksException {
-        byte s = AInt8SerializerDeserializer.getByte(outLeft.getBytes(), 1);
+        byte s = AInt8SerializerDeserializer.getByte(outLeft.getByteArray(), 1);
         switch (typeTag2) {
             case INT8: {
-                byte v2 = AInt8SerializerDeserializer.getByte(outRight.getBytes(), 1);
+                byte v2 = AInt8SerializerDeserializer.getByte(outRight.getByteArray(), 1);
                 return compareByte(s, v2);
             }
             case INT16: {
-                short v2 = AInt16SerializerDeserializer.getShort(outRight.getBytes(), 1);
+                short v2 = AInt16SerializerDeserializer.getShort(outRight.getByteArray(), 1);
                 return compareShort(s, v2);
             }
             case INT32: {
-                int v2 = AInt32SerializerDeserializer.getInt(outRight.getBytes(), 1);
+                int v2 = AInt32SerializerDeserializer.getInt(outRight.getByteArray(), 1);
                 return compareInt(s, v2);
             }
             case INT64: {
-                long v2 = AInt64SerializerDeserializer.getLong(outRight.getBytes(), 1);
+                long v2 = AInt64SerializerDeserializer.getLong(outRight.getByteArray(), 1);
                 return compareLong(s, v2);
             }
             case FLOAT: {
-                float v2 = AFloatSerializerDeserializer.getFloat(outRight.getBytes(), 1);
+                float v2 = AFloatSerializerDeserializer.getFloat(outRight.getByteArray(), 1);
                 return compareFloat(s, v2);
             }
             case DOUBLE: {
-                double v2 = ADoubleSerializerDeserializer.getDouble(outRight.getBytes(), 1);
+                double v2 = ADoubleSerializerDeserializer.getDouble(outRight.getByteArray(), 1);
                 return compareDouble(s, v2);
             }
             default: {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ABooleanConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ABooleanConstructorDescriptor.java
index 701c20a4..c379b58 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ABooleanConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ABooleanConstructorDescriptor.java
@@ -66,7 +66,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 if (utf8BinaryComparator.compare(serString, 1, outInput.getLength(), TRUE, 0, 6) == 0) {
                                     booleanSerde.serialize(ABoolean.TRUE, out);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ACircleConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ACircleConstructorDescriptor.java
index 3e64b56..57e6f22 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ACircleConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ACircleConstructorDescriptor.java
@@ -64,7 +64,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 String s = new String(serString, 3, outInput.getLength() - 3, "UTF-8");
                                 int commaIndex = s.indexOf(',');
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADateConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADateConstructorDescriptor.java
index 83c9167..ab32e1e 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADateConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADateConstructorDescriptor.java
@@ -66,7 +66,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 offset = 3;
 
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADateTimeConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADateTimeConstructorDescriptor.java
index 2f5a203..1e53cf2 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADateTimeConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADateTimeConstructorDescriptor.java
@@ -67,7 +67,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
 
                                 offset = 3;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADoubleConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADoubleConstructorDescriptor.java
index 6a89def..271ae78 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADoubleConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADoubleConstructorDescriptor.java
@@ -77,7 +77,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
 
                                 if (utf8BinaryComparator
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADurationConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADurationConstructorDescriptor.java
index b4ded9e..1d3760f 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADurationConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ADurationConstructorDescriptor.java
@@ -67,7 +67,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 offset = 3;
                                 if (serString[offset] == '-') {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AFloatConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AFloatConstructorDescriptor.java
index 6cb6306..53a823c 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AFloatConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AFloatConstructorDescriptor.java
@@ -76,7 +76,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 if (utf8BinaryComparator
                                         .compare(serString, 1, outInput.getLength(), POSITIVE_INF, 0, 5) == 0) {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt16ConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt16ConstructorDescriptor.java
index c1ac181..ce0bc25 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt16ConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt16ConstructorDescriptor.java
@@ -65,7 +65,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 offset = 3;
                                 value = 0;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt32ConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt32ConstructorDescriptor.java
index 2edfbdd..f3e2faf 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt32ConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt32ConstructorDescriptor.java
@@ -64,7 +64,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 offset = 3;
                                 value = 0;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt64ConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt64ConstructorDescriptor.java
index 5fd7a3a..a307da2 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt64ConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt64ConstructorDescriptor.java
@@ -65,7 +65,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 offset = 3;
                                 value = 0;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt8ConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt8ConstructorDescriptor.java
index 6719fbf..983ffe7 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt8ConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AInt8ConstructorDescriptor.java
@@ -65,7 +65,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 offset = 3;
                                 value = 0;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ALineConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ALineConstructorDescriptor.java
index 87063c3..2cc090a 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ALineConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ALineConstructorDescriptor.java
@@ -64,7 +64,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 String s = new String(serString, 3, outInput.getLength() - 3, "UTF-8");
                                 int commaIndex = s.indexOf(',');
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ANullConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ANullConstructorDescriptor.java
index 963e95d..a94d581 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ANullConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ANullConstructorDescriptor.java
@@ -59,7 +59,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 if (utf8BinaryComparator.compare(serString, 1, outInput.getLength(), NULL, 0, 6) == 0) {
                                     nullSerde.serialize(ANull.NULL, out);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/APoint3DConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/APoint3DConstructorDescriptor.java
index 1c0962c..5169362 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/APoint3DConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/APoint3DConstructorDescriptor.java
@@ -63,7 +63,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 String s = new String(serString, 3, outInput.getLength() - 3, "UTF-8");
                                 int firstCommaIndex = s.indexOf(',');
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/APointConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/APointConstructorDescriptor.java
index 378e619..22c89dc 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/APointConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/APointConstructorDescriptor.java
@@ -62,7 +62,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 String s = new String(serString, 3, outInput.getLength() - 3, "UTF-8");
                                 aPoint.setValue(Double.parseDouble(s.substring(0, s.indexOf(','))),
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/APolygonConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/APolygonConstructorDescriptor.java
index 8a88c0e..6511b2d 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/APolygonConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/APolygonConstructorDescriptor.java
@@ -59,7 +59,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 String s = new String(serString, 3, outInput.getLength() - 3, "UTF-8");
                                 String[] points = s.split(" ");
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ARectangleConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ARectangleConstructorDescriptor.java
index c2b98f9..4ffd671 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ARectangleConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ARectangleConstructorDescriptor.java
@@ -65,7 +65,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 String s = new String(serString, 3, outInput.getLength() - 3, "UTF-8");
                                 int commaIndex = s.indexOf(',');
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AStringConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AStringConstructorDescriptor.java
index 982871d..11aa889 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AStringConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/AStringConstructorDescriptor.java
@@ -56,9 +56,9 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
-                                out.write(outInput.getBytes(), outInput.getStartIndex(), outInput.getLength());
+                                out.write(outInput.getByteArray(), outInput.getStartOffset(), outInput.getLength());
                             } else if (serString[0] == SER_NULL_TYPE_TAG)
                                 nullSerde.serialize(ANull.NULL, out);
                             else
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ATimeConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ATimeConstructorDescriptor.java
index 10b6af5..46f0738 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ATimeConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/constructors/ATimeConstructorDescriptor.java
@@ -66,7 +66,7 @@
                         try {
                             outInput.reset();
                             eval.evaluate(tuple);
-                            byte[] serString = outInput.getBytes();
+                            byte[] serString = outInput.getByteArray();
                             if (serString[0] == SER_STRING_TYPE_TAG) {
                                 offset = 3;
                                 if (serString[offset + 2] != ':' || serString[offset + 5] != ':')
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/AbstractStringContainsEval.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/AbstractStringContainsEval.java
index b0f4580..9225f6a 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/AbstractStringContainsEval.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/AbstractStringContainsEval.java
@@ -39,8 +39,8 @@
         evalPattern.evaluate(tuple);
         array0.reset();
         evalString.evaluate(tuple);
-        byte[] b1 = array0.getBytes();
-        byte[] b2 = array1.getBytes();
+        byte[] b1 = array0.getByteArray();
+        byte[] b2 = array1.getByteArray();
         ABoolean res = findMatch(b1, b2) ? ABoolean.TRUE : ABoolean.FALSE;
         try {
             boolSerde.serialize(res, dout);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/AndDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/AndDescriptor.java
index 7e2877c..6a46c6d 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/AndDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/AndDescriptor.java
@@ -70,11 +70,11 @@
                             for (int i = 0; i < n; i++) {
                                 argOut.reset();
                                 evals[i].evaluate(tuple);
-                                if (argOut.getBytes()[0] == SER_NULL_TYPE_TAG) {
+                                if (argOut.getByteArray()[0] == SER_NULL_TYPE_TAG) {
                                     metNull = true;
                                     continue;
                                 }
-                                boolean argResult = ABooleanSerializerDeserializer.getBoolean(argOut.getBytes(), 1);
+                                boolean argResult = ABooleanSerializerDeserializer.getBoolean(argOut.getByteArray(), 1);
                                 res = res && argResult;
                             }
                             if (metNull) {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/AnyCollectionMemberDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/AnyCollectionMemberDescriptor.java
index 055d94b..aa8b430 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/AnyCollectionMemberDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/AnyCollectionMemberDescriptor.java
@@ -81,7 +81,7 @@
                     try {
                         outInputList.reset();
                         evalList.evaluate(tuple);
-                        byte[] serList = outInputList.getBytes();
+                        byte[] serList = outInputList.getByteArray();
 
                         if (serList[0] == SER_NULL_TYPE_TAG) {
                             nullSerde.serialize(ANull.NULL, out);
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 c82bb9b..32498c3 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
@@ -7,10 +7,10 @@
 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.base.IBinaryAccessor;
-import edu.uci.ics.asterix.runtime.accessors.cast.ACastVisitor;
 import edu.uci.ics.asterix.runtime.evaluators.base.AbstractScalarFunctionDynamicDescriptor;
+import edu.uci.ics.asterix.runtime.pointables.PointableAllocator;
+import edu.uci.ics.asterix.runtime.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.runtime.pointables.cast.ACastVisitor;
 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;
@@ -58,10 +58,12 @@
                 final IEvaluator recEvaluator = recordEvalFactory.createEvaluator(recordBuffer);
 
                 return new IEvaluator() {
-                    final ARecordAccessor recAccessor = new ARecordAccessor(inputType);
-                    final ARecordAccessor resultAccessor = new ARecordAccessor(reqType);
+                    // pointable allocator
+                    private PointableAllocator allocator = new PointableAllocator();
+                    final IVisitablePointable recAccessor = allocator.allocateRecordValue(inputType);
+                    final IVisitablePointable resultAccessor = allocator.allocateRecordValue(reqType);
                     final ACastVisitor castVisitor = new ACastVisitor();
-                    final Triple<IBinaryAccessor, IAType, Boolean> arg = new Triple<IBinaryAccessor, IAType, Boolean>(
+                    final Triple<IVisitablePointable, IAType, Boolean> arg = new Triple<IVisitablePointable, IAType, Boolean>(
                             resultAccessor, reqType, Boolean.FALSE);
 
                     @Override
@@ -69,10 +71,9 @@
                         try {
                             recordBuffer.reset();
                             recEvaluator.evaluate(tuple);
-                            recAccessor.reset(recordBuffer.getBytes(), recordBuffer.getStartIndex(),
-                                    recordBuffer.getLength());
+                            recAccessor.set(recordBuffer);
                             recAccessor.accept(castVisitor, arg);
-                            out.write(resultAccessor.getBytes(), resultAccessor.getStartIndex(),
+                            out.write(resultAccessor.getByteArray(), resultAccessor.getStartOffset(),
                                     resultAccessor.getLength());
                         } catch (Exception ioe) {
                             throw new AlgebricksException(ioe);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreateCircleDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreateCircleDescriptor.java
index 463dfd2..76c2206 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreateCircleDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreateCircleDescriptor.java
@@ -64,11 +64,11 @@
                         eval1.evaluate(tuple);
 
                         try {
-                            aPoint.setValue(ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                            aPoint.setValue(ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                     APointSerializerDeserializer.getCoordinateOffset(Coordinate.X)),
-                                    ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                    ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                             APointSerializerDeserializer.getCoordinateOffset(Coordinate.Y)));
-                            aCircle.setValue(aPoint, ADoubleSerializerDeserializer.getDouble(outInput1.getBytes(), 1));
+                            aCircle.setValue(aPoint, ADoubleSerializerDeserializer.getDouble(outInput1.getByteArray(), 1));
                             circleSerde.serialize(aCircle, out);
                         } catch (IOException e1) {
                             throw new AlgebricksException(e1);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreateLineDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreateLineDescriptor.java
index fab060d..8ca81cd 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreateLineDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreateLineDescriptor.java
@@ -64,13 +64,13 @@
                         eval1.evaluate(tuple);
 
                         try {
-                            aPoint[0].setValue(ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                            aPoint[0].setValue(ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                     APointSerializerDeserializer.getCoordinateOffset(Coordinate.X)),
-                                    ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                    ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                             APointSerializerDeserializer.getCoordinateOffset(Coordinate.Y)));
-                            aPoint[1].setValue(ADoubleSerializerDeserializer.getDouble(outInput1.getBytes(),
+                            aPoint[1].setValue(ADoubleSerializerDeserializer.getDouble(outInput1.getByteArray(),
                                     APointSerializerDeserializer.getCoordinateOffset(Coordinate.X)),
-                                    ADoubleSerializerDeserializer.getDouble(outInput1.getBytes(),
+                                    ADoubleSerializerDeserializer.getDouble(outInput1.getByteArray(),
                                             APointSerializerDeserializer.getCoordinateOffset(Coordinate.Y)));
                             aLine.setValue(aPoint[0], aPoint[1]);
                             lineSerde.serialize(aLine, out);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreatePointDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreatePointDescriptor.java
index f7d284c..2a4a0f1 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreatePointDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreatePointDescriptor.java
@@ -60,8 +60,8 @@
                         eval1.evaluate(tuple);
 
                         try {
-                            aPoint.setValue(ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(), 1),
-                                    ADoubleSerializerDeserializer.getDouble(outInput1.getBytes(), 1));
+                            aPoint.setValue(ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(), 1),
+                                    ADoubleSerializerDeserializer.getDouble(outInput1.getByteArray(), 1));
                             pointSerde.serialize(aPoint, out);
 
                         } catch (IOException e1) {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreatePolygonDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreatePolygonDescriptor.java
index e0b424f..eb3959e 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreatePolygonDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreatePolygonDescriptor.java
@@ -62,7 +62,7 @@
                             outInput.reset();
                             argEvals[i].evaluate(tuple);
                             try {
-                                out.write(outInput.getBytes(), outInput.getStartIndex() + 1, outInput.getLength() - 1);
+                                out.write(outInput.getByteArray(), outInput.getStartOffset() + 1, outInput.getLength() - 1);
                             } catch (IOException e) {
                                 throw new AlgebricksException(e);
                             }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreateRectangleDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreateRectangleDescriptor.java
index 1e4a98a..9d7bb5f 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreateRectangleDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/CreateRectangleDescriptor.java
@@ -63,13 +63,13 @@
                         eval1.evaluate(tuple);
 
                         try {
-                            aPoint[0].setValue(ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                            aPoint[0].setValue(ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                     APointSerializerDeserializer.getCoordinateOffset(Coordinate.X)),
-                                    ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                    ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                             APointSerializerDeserializer.getCoordinateOffset(Coordinate.Y)));
-                            aPoint[1].setValue(ADoubleSerializerDeserializer.getDouble(outInput1.getBytes(),
+                            aPoint[1].setValue(ADoubleSerializerDeserializer.getDouble(outInput1.getByteArray(),
                                     APointSerializerDeserializer.getCoordinateOffset(Coordinate.X)),
-                                    ADoubleSerializerDeserializer.getDouble(outInput1.getBytes(),
+                                    ADoubleSerializerDeserializer.getDouble(outInput1.getByteArray(),
                                             APointSerializerDeserializer.getCoordinateOffset(Coordinate.Y)));
                             if (aPoint[0].getX() > aPoint[1].getX() || aPoint[0].getY() > aPoint[1].getY()) {
                                 throw new IllegalArgumentException(
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/EditDistanceCheckDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/EditDistanceCheckDescriptor.java
index cb89147..550cfc4 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/EditDistanceCheckDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/EditDistanceCheckDescriptor.java
@@ -75,7 +75,7 @@
             super.runArgEvals(tuple);
             int edThreshStart = argOut.getLength();
             edThreshEval.evaluate(tuple);
-            edThresh = IntegerSerializerDeserializer.getInt(argOut.getBytes(), edThreshStart + typeIndicatorSize);
+            edThresh = IntegerSerializerDeserializer.getInt(argOut.getByteArray(), edThreshStart + typeIndicatorSize);
         }
 
         @Override
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/FieldAccessByNameDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/FieldAccessByNameDescriptor.java
index 67a8dfa..a0b4af0 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/FieldAccessByNameDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/FieldAccessByNameDescriptor.java
@@ -85,7 +85,7 @@
                         eval0.evaluate(tuple);
                         outInput1.reset();
                         eval1.evaluate(tuple);
-                        byte[] serRecord = outInput0.getBytes();
+                        byte[] serRecord = outInput0.getByteArray();
 
                         if (serRecord[0] == SER_NULL_TYPE_TAG) {
                             nullSerde.serialize(ANull.NULL, out);
@@ -97,7 +97,7 @@
                                     + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(serRecord[0]));
                         }
 
-                        byte[] serFldName = outInput1.getBytes();
+                        byte[] serFldName = outInput1.getByteArray();
                         fieldValueOffset = ARecordSerializerDeserializer.getFieldOffsetByName(serRecord, serFldName);
                         if (fieldValueOffset < 0) {
                             out.writeByte(ATypeTag.NULL.serialize());
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/GetItemDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/GetItemDescriptor.java
index 4ce843b..34ea593 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/GetItemDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/GetItemDescriptor.java
@@ -87,7 +87,7 @@
                         evalList.evaluate(tuple);
                         outInputIdx.reset();
                         evalIdx.evaluate(tuple);
-                        byte[] serOrderedList = outInputList.getBytes();
+                        byte[] serOrderedList = outInputList.getByteArray();
 
                         if (serOrderedList[0] == SER_NULL_TYPE_TAG) {
                             nullSerde.serialize(ANull.NULL, out);
@@ -99,7 +99,7 @@
                                     + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(serOrderedList[0]));
                         }
 
-                        itemIndex = IntegerSerializerDeserializer.getInt(outInputIdx.getBytes(), 1);
+                        itemIndex = IntegerSerializerDeserializer.getInt(outInputIdx.getByteArray(), 1);
                         if (itemIndex >= AOrderedListSerializerDeserializer.getNumberOfItems(serOrderedList)) {
                             out.writeByte(SER_NULL_TYPE_TAG);
                             return;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/InjectFailureDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/InjectFailureDescriptor.java
index 478316f..941bd53 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/InjectFailureDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/InjectFailureDescriptor.java
@@ -55,9 +55,9 @@
                             // evaluator the failure condition
                             argOut.reset();
                             evals[1].evaluate(tuple);
-                            ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getBytes()[0]);
+                            ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getByteArray()[0]);
                             if (typeTag == ATypeTag.BOOLEAN) {
-                                boolean argResult = ABooleanSerializerDeserializer.getBoolean(argOut.getBytes(), 1);
+                                boolean argResult = ABooleanSerializerDeserializer.getBoolean(argOut.getByteArray(), 1);
                                 if (argResult)
                                     throw new AlgebricksException("Injecting a intended failure");
                             }
@@ -65,7 +65,7 @@
                             // evaluate the real evaluator
                             argOut.reset();
                             evals[0].evaluate(tuple);
-                            output.getDataOutput().write(argOut.getBytes(), argOut.getStartIndex(), argOut.getLength());
+                            output.getDataOutput().write(argOut.getByteArray(), argOut.getStartOffset(), argOut.getLength());
                         } catch (IOException e) {
                             throw new AlgebricksException(e);
                         }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/IsNullDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/IsNullDescriptor.java
index 59bce2d..9583c75 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/IsNullDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/IsNullDescriptor.java
@@ -49,7 +49,7 @@
                     public void evaluate(IFrameTupleReference tuple) throws AlgebricksException {
                         argOut.reset();
                         eval.evaluate(tuple);
-                        boolean isNull = argOut.getBytes()[argOut.getStartIndex()] == SER_NULL_TYPE_TAG;
+                        boolean isNull = argOut.getByteArray()[argOut.getStartOffset()] == SER_NULL_TYPE_TAG;
                         ABoolean res = isNull ? ABoolean.TRUE : ABoolean.FALSE;
                         try {
                             AObjectSerializerDeserializer.INSTANCE.serialize(res, out);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/LenDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/LenDescriptor.java
index 769fdb7..b87eb37 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/LenDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/LenDescriptor.java
@@ -67,7 +67,7 @@
                         inputVal.reset();
                         evalList.evaluate(tuple);
 
-                        byte[] serList = inputVal.getBytes();
+                        byte[] serList = inputVal.getByteArray();
 
                         if (serList[0] == SER_NULL_TYPE_TAG) {
                             try {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/LikeDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/LikeDescriptor.java
index 1090742..3a4af19 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/LikeDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/LikeDescriptor.java
@@ -91,7 +91,7 @@
                         try {
                             array0.reset();
                             evalPattern.evaluate(tuple);
-                            if (array0.getBytes()[0] == SER_NULL_TYPE_TAG) {
+                            if (array0.getByteArray()[0] == SER_NULL_TYPE_TAG) {
                                 nullSerde.serialize(ANull.NULL, dout);
                                 return;
                             }
@@ -100,7 +100,7 @@
                                 first = false;
                                 newPattern = true;
                             } else {
-                                int c = strComp.compare(array0.getBytes(), array0.getStartIndex(), array0.getLength(),
+                                int c = strComp.compare(array0.getByteArray(), array0.getStartOffset(), array0.getLength(),
                                         lastPattern.getByteArray(), 0, lastPattern.size());
                                 if (c != 0) {
                                     newPattern = true;
@@ -108,7 +108,7 @@
                             }
                             if (newPattern) {
                                 lastPattern.reset();
-                                lastPattern.write(array0.getBytes(), array0.getStartIndex(), array0.getLength());
+                                lastPattern.write(array0.getByteArray(), array0.getStartOffset(), array0.getLength());
                                 // ! object creation !
                                 DataInputStream di = new DataInputStream(new ByteArrayInputStream(
                                         lastPattern.getByteArray()));
@@ -118,7 +118,7 @@
                             }
                             array0.reset();
                             evalString.evaluate(tuple);
-                            if (array0.getBytes()[0] == SER_NULL_TYPE_TAG) {
+                            if (array0.getByteArray()[0] == SER_NULL_TYPE_TAG) {
                                 nullSerde.serialize(ANull.NULL, dout);
                                 return;
                             }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NotDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NotDescriptor.java
index 1e4cb3e..0ddf9f6 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NotDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NotDescriptor.java
@@ -68,11 +68,11 @@
                         argOut.reset();
                         eval.evaluate(tuple);
                         try {
-                            if (argOut.getBytes()[0] == SER_BOOLEAN_TYPE_TAG) {
-                                boolean argRes = ABooleanSerializerDeserializer.getBoolean(argOut.getBytes(), 1);
+                            if (argOut.getByteArray()[0] == SER_BOOLEAN_TYPE_TAG) {
+                                boolean argRes = ABooleanSerializerDeserializer.getBoolean(argOut.getByteArray(), 1);
                                 ABoolean aResult = argRes ? (ABoolean.FALSE) : (ABoolean.TRUE);
                                 booleanSerde.serialize(aResult, out);
-                            } else if (argOut.getBytes()[0] == SER_NULL_TYPE_TAG)
+                            } else if (argOut.getByteArray()[0] == SER_NULL_TYPE_TAG)
                                 nullSerde.serialize(ANull.NULL, out);
                             else
                                 throw new AlgebricksException(errorMessage);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericAddDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericAddDescriptor.java
index 405a006..b525bc3 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericAddDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericAddDescriptor.java
@@ -88,36 +88,36 @@
                                     evalLeft.evaluate(tuple);
                                 else
                                     evalRight.evaluate(tuple);
-                                typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getBytes()[0]);
+                                typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getByteArray()[0]);
                                 switch (typeTag) {
                                     case INT8: {
                                         metInt8 = true;
-                                        operands[i] = AInt8SerializerDeserializer.getByte(argOut.getBytes(), 1);
+                                        operands[i] = AInt8SerializerDeserializer.getByte(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case INT16: {
                                         metInt16 = true;
-                                        operands[i] = AInt16SerializerDeserializer.getShort(argOut.getBytes(), 1);
+                                        operands[i] = AInt16SerializerDeserializer.getShort(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case INT32: {
                                         metInt32 = true;
-                                        operands[i] = AInt32SerializerDeserializer.getInt(argOut.getBytes(), 1);
+                                        operands[i] = AInt32SerializerDeserializer.getInt(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case INT64: {
                                         metInt64 = true;
-                                        operands[i] = AInt64SerializerDeserializer.getLong(argOut.getBytes(), 1);
+                                        operands[i] = AInt64SerializerDeserializer.getLong(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case FLOAT: {
                                         metFloat = true;
-                                        operands[i] = AFloatSerializerDeserializer.getFloat(argOut.getBytes(), 1);
+                                        operands[i] = AFloatSerializerDeserializer.getFloat(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case DOUBLE: {
                                         metDouble = true;
-                                        operands[i] = ADoubleSerializerDeserializer.getDouble(argOut.getBytes(), 1);
+                                        operands[i] = ADoubleSerializerDeserializer.getDouble(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case NULL: {
@@ -131,7 +131,7 @@
                                                 : "Right"
                                                         + " Operand of Addition can not be "
                                                         + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut
-                                                                .getBytes()[0]));
+                                                                .getByteArray()[0]));
                                     }
                                 }
                             }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericDivideDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericDivideDescriptor.java
index a6d0f47..39c1fbf 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericDivideDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericDivideDescriptor.java
@@ -88,36 +88,36 @@
                                     evalLeft.evaluate(tuple);
                                 else
                                     evalRight.evaluate(tuple);
-                                typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getBytes()[0]);
+                                typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getByteArray()[0]);
                                 switch (typeTag) {
                                     case INT8: {
                                         metInt8 = true;
-                                        operands[i] = AInt8SerializerDeserializer.getByte(argOut.getBytes(), 1);
+                                        operands[i] = AInt8SerializerDeserializer.getByte(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case INT16: {
                                         metInt16 = true;
-                                        operands[i] = AInt16SerializerDeserializer.getShort(argOut.getBytes(), 1);
+                                        operands[i] = AInt16SerializerDeserializer.getShort(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case INT32: {
                                         metInt32 = true;
-                                        operands[i] = AInt32SerializerDeserializer.getInt(argOut.getBytes(), 1);
+                                        operands[i] = AInt32SerializerDeserializer.getInt(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case INT64: {
                                         metInt64 = true;
-                                        operands[i] = AInt64SerializerDeserializer.getLong(argOut.getBytes(), 1);
+                                        operands[i] = AInt64SerializerDeserializer.getLong(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case FLOAT: {
                                         metFloat = true;
-                                        operands[i] = AFloatSerializerDeserializer.getFloat(argOut.getBytes(), 1);
+                                        operands[i] = AFloatSerializerDeserializer.getFloat(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case DOUBLE: {
                                         metDouble = true;
-                                        operands[i] = ADoubleSerializerDeserializer.getDouble(argOut.getBytes(), 1);
+                                        operands[i] = ADoubleSerializerDeserializer.getDouble(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case NULL: {
@@ -131,7 +131,7 @@
                                                 : "Right"
                                                         + " Operand of Division can not be "
                                                         + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut
-                                                                .getBytes()[0]));
+                                                                .getByteArray()[0]));
                                     }
                                 }
                             }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericMultiplyDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericMultiplyDescriptor.java
index 33a1a90..ed7626e 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericMultiplyDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericMultiplyDescriptor.java
@@ -88,36 +88,36 @@
                                     evalLeft.evaluate(tuple);
                                 else
                                     evalRight.evaluate(tuple);
-                                typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getBytes()[0]);
+                                typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getByteArray()[0]);
                                 switch (typeTag) {
                                     case INT8: {
                                         metInt8 = true;
-                                        operands[i] = AInt8SerializerDeserializer.getByte(argOut.getBytes(), 1);
+                                        operands[i] = AInt8SerializerDeserializer.getByte(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case INT16: {
                                         metInt16 = true;
-                                        operands[i] = AInt16SerializerDeserializer.getShort(argOut.getBytes(), 1);
+                                        operands[i] = AInt16SerializerDeserializer.getShort(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case INT32: {
                                         metInt32 = true;
-                                        operands[i] = AInt32SerializerDeserializer.getInt(argOut.getBytes(), 1);
+                                        operands[i] = AInt32SerializerDeserializer.getInt(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case INT64: {
                                         metInt64 = true;
-                                        operands[i] = AInt64SerializerDeserializer.getLong(argOut.getBytes(), 1);
+                                        operands[i] = AInt64SerializerDeserializer.getLong(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case FLOAT: {
                                         metFloat = true;
-                                        operands[i] = AFloatSerializerDeserializer.getFloat(argOut.getBytes(), 1);
+                                        operands[i] = AFloatSerializerDeserializer.getFloat(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case DOUBLE: {
                                         metDouble = true;
-                                        operands[i] = ADoubleSerializerDeserializer.getDouble(argOut.getBytes(), 1);
+                                        operands[i] = ADoubleSerializerDeserializer.getDouble(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case NULL: {
@@ -131,7 +131,7 @@
                                                 : "Right"
                                                         + " Operand of Multiplication can not be "
                                                         + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut
-                                                                .getBytes()[0]));
+                                                                .getByteArray()[0]));
                                     }
                                 }
                             }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericSubtractDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericSubtractDescriptor.java
index ffed14e..77ae37c 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericSubtractDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericSubtractDescriptor.java
@@ -88,36 +88,36 @@
                                     evalLeft.evaluate(tuple);
                                 else
                                     evalRight.evaluate(tuple);
-                                typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getBytes()[0]);
+                                typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getByteArray()[0]);
                                 switch (typeTag) {
                                     case INT8: {
                                         metInt8 = true;
-                                        operands[i] = AInt8SerializerDeserializer.getByte(argOut.getBytes(), 1);
+                                        operands[i] = AInt8SerializerDeserializer.getByte(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case INT16: {
                                         metInt16 = true;
-                                        operands[i] = AInt16SerializerDeserializer.getShort(argOut.getBytes(), 1);
+                                        operands[i] = AInt16SerializerDeserializer.getShort(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case INT32: {
                                         metInt32 = true;
-                                        operands[i] = AInt32SerializerDeserializer.getInt(argOut.getBytes(), 1);
+                                        operands[i] = AInt32SerializerDeserializer.getInt(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case INT64: {
                                         metInt64 = true;
-                                        operands[i] = AInt64SerializerDeserializer.getLong(argOut.getBytes(), 1);
+                                        operands[i] = AInt64SerializerDeserializer.getLong(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case FLOAT: {
                                         metFloat = true;
-                                        operands[i] = AFloatSerializerDeserializer.getFloat(argOut.getBytes(), 1);
+                                        operands[i] = AFloatSerializerDeserializer.getFloat(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case DOUBLE: {
                                         metDouble = true;
-                                        operands[i] = ADoubleSerializerDeserializer.getDouble(argOut.getBytes(), 1);
+                                        operands[i] = ADoubleSerializerDeserializer.getDouble(argOut.getByteArray(), 1);
                                         break;
                                     }
                                     case NULL: {
@@ -131,7 +131,7 @@
                                                 : "Right"
                                                         + " Operand of Substraction can not be "
                                                         + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut
-                                                                .getBytes()[0]));
+                                                                .getByteArray()[0]));
                                     }
                                 }
                             }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericUnaryMinusDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericUnaryMinusDescriptor.java
index cd234cb..73181d6 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericUnaryMinusDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/NumericUnaryMinusDescriptor.java
@@ -80,44 +80,44 @@
                         argOut.reset();
                         eval.evaluate(tuple);
                         try {
-                            if (argOut.getBytes()[0] == serNullTypeTag) {
+                            if (argOut.getByteArray()[0] == serNullTypeTag) {
                                 serde = AqlSerializerDeserializerProvider.INSTANCE
                                         .getSerializerDeserializer(BuiltinType.ANULL);
                                 serde.serialize(ANull.NULL, out);
                                 return;
-                            } else if (argOut.getBytes()[0] == serInt8TypeTag) {
+                            } else if (argOut.getByteArray()[0] == serInt8TypeTag) {
                                 serde = AqlSerializerDeserializerProvider.INSTANCE
                                         .getSerializerDeserializer(BuiltinType.AINT8);
-                                aInt8.setValue((byte) -AInt8SerializerDeserializer.getByte(argOut.getBytes(), 1));
+                                aInt8.setValue((byte) -AInt8SerializerDeserializer.getByte(argOut.getByteArray(), 1));
                                 serde.serialize(aInt8, out);
-                            } else if (argOut.getBytes()[0] == serInt16TypeTag) {
+                            } else if (argOut.getByteArray()[0] == serInt16TypeTag) {
                                 serde = AqlSerializerDeserializerProvider.INSTANCE
                                         .getSerializerDeserializer(BuiltinType.AINT16);
-                                aInt16.setValue((short) -AInt16SerializerDeserializer.getShort(argOut.getBytes(), 1));
+                                aInt16.setValue((short) -AInt16SerializerDeserializer.getShort(argOut.getByteArray(), 1));
                                 serde.serialize(aInt16, out);
-                            } else if (argOut.getBytes()[0] == serInt32TypeTag) {
+                            } else if (argOut.getByteArray()[0] == serInt32TypeTag) {
                                 serde = AqlSerializerDeserializerProvider.INSTANCE
                                         .getSerializerDeserializer(BuiltinType.AINT32);
-                                aInt32.setValue(-AInt32SerializerDeserializer.getInt(argOut.getBytes(), 1));
+                                aInt32.setValue(-AInt32SerializerDeserializer.getInt(argOut.getByteArray(), 1));
                                 serde.serialize(aInt32, out);
-                            } else if (argOut.getBytes()[0] == serInt64TypeTag) {
+                            } else if (argOut.getByteArray()[0] == serInt64TypeTag) {
                                 serde = AqlSerializerDeserializerProvider.INSTANCE
                                         .getSerializerDeserializer(BuiltinType.AINT64);
-                                aInt64.setValue(-AInt64SerializerDeserializer.getLong(argOut.getBytes(), 1));
+                                aInt64.setValue(-AInt64SerializerDeserializer.getLong(argOut.getByteArray(), 1));
                                 serde.serialize(aInt64, out);
-                            } else if (argOut.getBytes()[0] == serFloatTypeTag) {
+                            } else if (argOut.getByteArray()[0] == serFloatTypeTag) {
                                 serde = AqlSerializerDeserializerProvider.INSTANCE
                                         .getSerializerDeserializer(BuiltinType.AFLOAT);
-                                aFloat.setValue(-AFloatSerializerDeserializer.getFloat(argOut.getBytes(), 1));
+                                aFloat.setValue(-AFloatSerializerDeserializer.getFloat(argOut.getByteArray(), 1));
                                 serde.serialize(aFloat, out);
-                            } else if (argOut.getBytes()[0] == serDoubleTypeTag) {
+                            } else if (argOut.getByteArray()[0] == serDoubleTypeTag) {
                                 serde = AqlSerializerDeserializerProvider.INSTANCE
                                         .getSerializerDeserializer(BuiltinType.ADOUBLE);
-                                aDouble.setValue(-ADoubleSerializerDeserializer.getDouble(argOut.getBytes(), 1));
+                                aDouble.setValue(-ADoubleSerializerDeserializer.getDouble(argOut.getByteArray(), 1));
                                 serde.serialize(aDouble, out);
                             } else {
                                 throw new NotImplementedException("Unary minus is not implemented for "
-                                        + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getBytes()[0]));
+                                        + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argOut.getByteArray()[0]));
                             }
                         } catch (HyracksDataException e) {
                             throw new AlgebricksException(e);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/OpenRecordConstructorDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/OpenRecordConstructorDescriptor.java
index aeb6b15..cc53a04 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/OpenRecordConstructorDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/OpenRecordConstructorDescriptor.java
@@ -81,7 +81,7 @@
                                     evalNames[i].evaluate(tuple);
                                     recBuilder.addField(fieldNameBuffer, fieldValueBuffer);
                                 } else {
-                                    if (fieldValueBuffer.getBytes()[0] != ATypeTag.NULL.serialize()) {
+                                    if (fieldValueBuffer.getByteArray()[0] != ATypeTag.NULL.serialize()) {
                                         recBuilder.addField(closedFieldId, fieldValueBuffer);
                                     }
                                     closedFieldId++;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/OrDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/OrDescriptor.java
index 837d54a..c310164 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/OrDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/OrDescriptor.java
@@ -69,11 +69,11 @@
                             for (int i = 0; i < n; i++) {
                                 argOut.reset();
                                 evals[i].evaluate(tuple);
-                                if (argOut.getBytes()[0] == SER_NULL_TYPE_TAG) {
+                                if (argOut.getByteArray()[0] == SER_NULL_TYPE_TAG) {
                                     metNull = true;
                                     continue;
                                 }
-                                boolean argResult = ABooleanSerializerDeserializer.getBoolean(argOut.getBytes(), 1);
+                                boolean argResult = ABooleanSerializerDeserializer.getBoolean(argOut.getByteArray(), 1);
                                 if (argResult == true) {
                                     res = true;
                                     break;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/PrefixLenDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/PrefixLenDescriptor.java
index 880e84e..3e70f3c 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/PrefixLenDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/PrefixLenDescriptor.java
@@ -65,19 +65,19 @@
                         // length
                         inputVal.reset();
                         evalLen.evaluate(tuple);
-                        int length = IntegerSerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                        int length = IntegerSerializerDeserializer.getInt(inputVal.getByteArray(), 1);
 
                         // similarity threshold
                         inputVal.reset();
                         evalThreshold.evaluate(tuple);
                         float similarityThreshold = (float) ADoubleSerializerDeserializer.getDouble(
-                                inputVal.getBytes(), 1);
+                                inputVal.getByteArray(), 1);
 
                         // similarity name
                         inputVal.reset();
                         evalSimilarity.evaluate(tuple);
                         SimilarityFilters similarityFilters = similarityFiltersCache.get(similarityThreshold,
-                                inputVal.getBytes());
+                                inputVal.getByteArray());
 
                         int prefixLength = similarityFilters.getPrefixLength(length);
                         res.setValue(prefixLength);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/PrefixLenJaccardDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/PrefixLenJaccardDescriptor.java
index ab95a36..8021c2d 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/PrefixLenJaccardDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/PrefixLenJaccardDescriptor.java
@@ -64,12 +64,12 @@
                         // length
                         inputVal.reset();
                         evalLen.evaluate(tuple);
-                        int length = IntegerSerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                        int length = IntegerSerializerDeserializer.getInt(inputVal.getByteArray(), 1);
 
                         // similarity threshold
                         inputVal.reset();
                         evalThreshold.evaluate(tuple);
-                        float similarityThreshold = (float) AFloatSerializerDeserializer.getFloat(inputVal.getBytes(),
+                        float similarityThreshold = (float) AFloatSerializerDeserializer.getFloat(inputVal.getByteArray(),
                                 1);
 
                         if (similarityThreshold != similarityThresholdCache || similarityFilters == null) {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/RegExpDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/RegExpDescriptor.java
index 2a415e4..596439e 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/RegExpDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/RegExpDescriptor.java
@@ -91,7 +91,7 @@
                         try {
                             array0.reset();
                             evalPattern.evaluate(tuple);
-                            if (array0.getBytes()[0] == SER_NULL_TYPE_TAG) {
+                            if (array0.getByteArray()[0] == SER_NULL_TYPE_TAG) {
                                 nullSerde.serialize(ANull.NULL, dout);
                                 return;
                             }
@@ -100,7 +100,7 @@
                                 first = false;
                                 newPattern = true;
                             } else {
-                                int c = strComp.compare(array0.getBytes(), array0.getStartIndex(), array0.getLength(),
+                                int c = strComp.compare(array0.getByteArray(), array0.getStartOffset(), array0.getLength(),
                                         lastPattern.getByteArray(), 0, lastPattern.size());
                                 if (c != 0) {
                                     newPattern = true;
@@ -108,7 +108,7 @@
                             }
                             if (newPattern) {
                                 lastPattern.reset();
-                                lastPattern.write(array0.getBytes(), array0.getStartIndex(), array0.getLength());
+                                lastPattern.write(array0.getByteArray(), array0.getStartOffset(), array0.getLength());
                                 // ! object creation !
                                 DataInputStream di = new DataInputStream(new ByteArrayInputStream(
                                         lastPattern.getByteArray()));
@@ -118,7 +118,7 @@
                             }
                             array0.reset();
                             evalString.evaluate(tuple);
-                            if (array0.getBytes()[0] == SER_NULL_TYPE_TAG) {
+                            if (array0.getByteArray()[0] == SER_NULL_TYPE_TAG) {
                                 nullSerde.serialize(ANull.NULL, dout);
                                 return;
                             }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SimilarityDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SimilarityDescriptor.java
index 85adfe1..90788a6 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SimilarityDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SimilarityDescriptor.java
@@ -85,21 +85,21 @@
                         inputVal.reset();
                         evalThreshold.evaluate(tuple);
                         float similarityThreshold = (float) ADoubleSerializerDeserializer.getDouble(
-                                inputVal.getBytes(), 1);
+                                inputVal.getByteArray(), 1);
 
                         // similarity name
                         inputVal.reset();
                         evalSimilarity.evaluate(tuple);
                         SimilarityFilters similarityFilters = similarityFiltersCache.get(similarityThreshold,
-                                inputVal.getBytes());
+                                inputVal.getByteArray());
 
                         inputVal.reset();
                         evalLen1.evaluate(tuple);
-                        int length1 = IntegerSerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                        int length1 = IntegerSerializerDeserializer.getInt(inputVal.getByteArray(), 1);
 
                         inputVal.reset();
                         evalLen2.evaluate(tuple);
-                        int length2 = IntegerSerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                        int length2 = IntegerSerializerDeserializer.getInt(inputVal.getByteArray(), 1);
 
                         float sim = 0;
 
@@ -114,7 +114,7 @@
                             inputVal.reset();
                             evalTokens1.evaluate(tuple);
 
-                            byte[] serList = inputVal.getBytes();
+                            byte[] serList = inputVal.getByteArray();
                             if (serList[0] != SER_ORDEREDLIST_TYPE_TAG && serList[0] != SER_UNORDEREDLIST_TYPE_TAG) {
                                 throw new AlgebricksException("Scan collection is not defined for values of type"
                                         + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(serList[0]));
@@ -123,7 +123,7 @@
                             int lengthTokens1;
                             if (serList[0] == SER_ORDEREDLIST_TYPE_TAG) {
                                 lengthTokens1 = AOrderedListSerializerDeserializer
-                                        .getNumberOfItems(inputVal.getBytes());
+                                        .getNumberOfItems(inputVal.getByteArray());
                                 // read tokens
                                 for (i = 0; i < lengthTokens1; i++) {
                                     int itemOffset;
@@ -136,7 +136,7 @@
                                 }
                             } else {
                                 lengthTokens1 = AUnorderedListSerializerDeserializer.getNumberOfItems(inputVal
-                                        .getBytes());
+                                        .getByteArray());
                                 // read tokens
                                 for (i = 0; i < lengthTokens1; i++) {
                                     int itemOffset;
@@ -158,7 +158,7 @@
                             inputVal.reset();
                             evalTokens2.evaluate(tuple);
 
-                            serList = inputVal.getBytes();
+                            serList = inputVal.getByteArray();
                             if (serList[0] != SER_ORDEREDLIST_TYPE_TAG && serList[0] != SER_UNORDEREDLIST_TYPE_TAG) {
                                 throw new AlgebricksException("Scan collection is not defined for values of type"
                                         + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(serList[0]));
@@ -167,7 +167,7 @@
                             int lengthTokens2;
                             if (serList[0] == SER_ORDEREDLIST_TYPE_TAG) {
                                 lengthTokens2 = AOrderedListSerializerDeserializer
-                                        .getNumberOfItems(inputVal.getBytes());
+                                        .getNumberOfItems(inputVal.getByteArray());
                                 // read tokens
                                 for (i = 0; i < lengthTokens2; i++) {
                                     int itemOffset;
@@ -180,7 +180,7 @@
                                 }
                             } else {
                                 lengthTokens2 = AUnorderedListSerializerDeserializer.getNumberOfItems(inputVal
-                                        .getBytes());
+                                        .getByteArray());
                                 // read tokens
                                 for (i = 0; i < lengthTokens2; i++) {
                                     int itemOffset;
@@ -200,7 +200,7 @@
                             // -- - token prefix - --
                             inputVal.reset();
                             evalTokenPrefix.evaluate(tuple);
-                            int tokenPrefix = IntegerSerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                            int tokenPrefix = IntegerSerializerDeserializer.getInt(inputVal.getByteArray(), 1);
 
                             //
                             // -- - position filter - --
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SimilarityJaccardCheckDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SimilarityJaccardCheckDescriptor.java
index 3021717..971bfee 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SimilarityJaccardCheckDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SimilarityJaccardCheckDescriptor.java
@@ -79,7 +79,7 @@
             super.runArgEvals(tuple);
             int jaccThreshStart = argOut.getLength();
             jaccThreshEval.evaluate(tuple);
-            jaccThresh = (float) AFloatSerializerDeserializer.getFloat(argOut.getBytes(), jaccThreshStart
+            jaccThresh = (float) AFloatSerializerDeserializer.getFloat(argOut.getByteArray(), jaccThreshStart
                     + typeIndicatorSize);
         }
 
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialAreaDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialAreaDescriptor.java
index 91496fd..7736fa7 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialAreaDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialAreaDescriptor.java
@@ -55,36 +55,36 @@
                         eval.evaluate(tuple);
 
                         try {
-                            byte[] bytes = argOut.getBytes();
+                            byte[] bytes = argOut.getByteArray();
                             ATypeTag tag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(bytes[0]);
                             double area = 0.0;
 
                             switch (tag) {
                                 case POLYGON:
-                                    int numOfPoints = AInt16SerializerDeserializer.getShort(argOut.getBytes(), 1);
+                                    int numOfPoints = AInt16SerializerDeserializer.getShort(argOut.getByteArray(), 1);
 
                                     if (numOfPoints < 3) {
                                         throw new AlgebricksException("Polygon must have at least 3 points");
                                     }
-                                    area = Math.abs(SpatialUtils.polygonArea(argOut.getBytes(), numOfPoints));
+                                    area = Math.abs(SpatialUtils.polygonArea(argOut.getByteArray(), numOfPoints));
                                     break;
                                 case CIRCLE:
-                                    double radius = ADoubleSerializerDeserializer.getDouble(argOut.getBytes(),
+                                    double radius = ADoubleSerializerDeserializer.getDouble(argOut.getByteArray(),
                                             ACircleSerializerDeserializer.getRadiusOffset());
                                     area = SpatialUtils.pi() * radius * radius;
                                     break;
                                 case RECTANGLE:
-                                    double x1 = ADoubleSerializerDeserializer.getDouble(argOut.getBytes(),
+                                    double x1 = ADoubleSerializerDeserializer.getDouble(argOut.getByteArray(),
                                             ARectangleSerializerDeserializer
                                                     .getBottomLeftCoordinateOffset(Coordinate.X));
-                                    double y1 = ADoubleSerializerDeserializer.getDouble(argOut.getBytes(),
+                                    double y1 = ADoubleSerializerDeserializer.getDouble(argOut.getByteArray(),
                                             ARectangleSerializerDeserializer
                                                     .getBottomLeftCoordinateOffset(Coordinate.Y));
 
-                                    double x2 = ADoubleSerializerDeserializer.getDouble(argOut.getBytes(),
+                                    double x2 = ADoubleSerializerDeserializer.getDouble(argOut.getByteArray(),
                                             ARectangleSerializerDeserializer
                                                     .getUpperRightCoordinateOffset(Coordinate.X));
-                                    double y2 = ADoubleSerializerDeserializer.getDouble(argOut.getBytes(),
+                                    double y2 = ADoubleSerializerDeserializer.getDouble(argOut.getByteArray(),
                                             ARectangleSerializerDeserializer
                                                     .getUpperRightCoordinateOffset(Coordinate.Y));
                                     area = (x2 - x1) * (y2 - y1);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialCellDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialCellDescriptor.java
index 85586d8..5182ee8 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialCellDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialCellDescriptor.java
@@ -75,20 +75,20 @@
                         eval3.evaluate(tuple);
 
                         try {
-                            ATypeTag tag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(outInput0.getBytes()[0]);
+                            ATypeTag tag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(outInput0.getByteArray()[0]);
                             if (tag == ATypeTag.POINT) {
-                                double xLoc = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                double xLoc = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                         APointSerializerDeserializer.getCoordinateOffset(Coordinate.X));
-                                double yLoc = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                double yLoc = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                         APointSerializerDeserializer.getCoordinateOffset(Coordinate.Y));
 
-                                double xOrigin = ADoubleSerializerDeserializer.getDouble(outInput1.getBytes(),
+                                double xOrigin = ADoubleSerializerDeserializer.getDouble(outInput1.getByteArray(),
                                         APointSerializerDeserializer.getCoordinateOffset(Coordinate.X));
-                                double yOrigin = ADoubleSerializerDeserializer.getDouble(outInput1.getBytes(),
+                                double yOrigin = ADoubleSerializerDeserializer.getDouble(outInput1.getByteArray(),
                                         APointSerializerDeserializer.getCoordinateOffset(Coordinate.Y));
 
-                                double xInc = ADoubleSerializerDeserializer.getDouble(outInput2.getBytes(), 1);
-                                double yInc = ADoubleSerializerDeserializer.getDouble(outInput3.getBytes(), 1);
+                                double xInc = ADoubleSerializerDeserializer.getDouble(outInput2.getByteArray(), 1);
+                                double yInc = ADoubleSerializerDeserializer.getDouble(outInput3.getByteArray(), 1);
 
                                 double x = xOrigin + (Math.floor((xLoc - xOrigin) / xInc)) * xInc;
                                 double y = yOrigin + (Math.floor((yLoc - yOrigin) / yInc)) * yInc;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialDistanceDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialDistanceDescriptor.java
index 3ea1994..3a474ca 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialDistanceDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialDistanceDescriptor.java
@@ -56,20 +56,20 @@
                         eval1.evaluate(tuple);
 
                         try {
-                            byte[] bytes0 = outInput0.getBytes();
-                            byte[] bytes1 = outInput1.getBytes();
+                            byte[] bytes0 = outInput0.getByteArray();
+                            byte[] bytes1 = outInput1.getByteArray();
                             ATypeTag tag0 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(bytes0[0]);
                             ATypeTag tag1 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(bytes1[0]);
                             double distance = 0.0;
                             if (tag0 == ATypeTag.POINT) {
                                 if (tag1 == ATypeTag.POINT) {
-                                    double x1 = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                    double x1 = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                             APointSerializerDeserializer.getCoordinateOffset(Coordinate.X));
-                                    double y1 = ADoubleSerializerDeserializer.getDouble(outInput0.getBytes(),
+                                    double y1 = ADoubleSerializerDeserializer.getDouble(outInput0.getByteArray(),
                                             APointSerializerDeserializer.getCoordinateOffset(Coordinate.Y));
-                                    double x2 = ADoubleSerializerDeserializer.getDouble(outInput1.getBytes(),
+                                    double x2 = ADoubleSerializerDeserializer.getDouble(outInput1.getByteArray(),
                                             APointSerializerDeserializer.getCoordinateOffset(Coordinate.X));
-                                    double y2 = ADoubleSerializerDeserializer.getDouble(outInput1.getBytes(),
+                                    double y2 = ADoubleSerializerDeserializer.getDouble(outInput1.getByteArray(),
                                             APointSerializerDeserializer.getCoordinateOffset(Coordinate.Y));
                                     distance = Math.sqrt(Math.pow(x2 - x1, 2) + Math.pow(y2 - y1, 2));
                                 } else {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialIntersectDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialIntersectDescriptor.java
index 272e7e0..07eeb10 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialIntersectDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SpatialIntersectDescriptor.java
@@ -751,20 +751,20 @@
 
                         try {
                             boolean res = false;
-                            ATypeTag tag0 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(outInput0.getBytes()[0]);
-                            ATypeTag tag1 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(outInput1.getBytes()[0]);
+                            ATypeTag tag0 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(outInput0.getByteArray()[0]);
+                            ATypeTag tag1 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(outInput1.getByteArray()[0]);
 
                             switch (tag0) {
                                 case POINT:
                                     switch (tag1) {
                                         case POINT:
-                                            if (ascDoubleComp.compare(outInput0.getBytes(),
+                                            if (ascDoubleComp.compare(outInput0.getByteArray(),
                                                     APointSerializerDeserializer.getCoordinateOffset(Coordinate.X), 8,
-                                                    outInput1.getBytes(),
+                                                    outInput1.getByteArray(),
                                                     APointSerializerDeserializer.getCoordinateOffset(Coordinate.X), 8) == 0) {
-                                                if (ascDoubleComp.compare(outInput0.getBytes(),
+                                                if (ascDoubleComp.compare(outInput0.getByteArray(),
                                                         APointSerializerDeserializer.getCoordinateOffset(Coordinate.Y),
-                                                        8, outInput1.getBytes(),
+                                                        8, outInput1.getByteArray(),
                                                         APointSerializerDeserializer.getCoordinateOffset(Coordinate.Y),
                                                         8) == 0) {
                                                     res = true;
@@ -772,16 +772,16 @@
                                             }
                                             break;
                                         case LINE:
-                                            res = pointOnLine(outInput1.getBytes(), outInput0.getBytes());
+                                            res = pointOnLine(outInput1.getByteArray(), outInput0.getByteArray());
                                             break;
                                         case POLYGON:
-                                            res = pointInPolygon(outInput0.getBytes(), outInput1.getBytes());
+                                            res = pointInPolygon(outInput0.getByteArray(), outInput1.getByteArray());
                                             break;
                                         case CIRCLE:
-                                            res = pointInCircle(outInput0.getBytes(), outInput1.getBytes());
+                                            res = pointInCircle(outInput0.getByteArray(), outInput1.getByteArray());
                                             break;
                                         case RECTANGLE:
-                                            res = pointInRectangle(outInput0.getBytes(), outInput1.getBytes());
+                                            res = pointInRectangle(outInput0.getByteArray(), outInput1.getByteArray());
                                             break;
                                         case NULL:
                                             res = false;
@@ -796,45 +796,45 @@
                                 case LINE:
                                     switch (tag1) {
                                         case POINT:
-                                            res = pointOnLine(outInput0.getBytes(), outInput1.getBytes());
+                                            res = pointOnLine(outInput0.getByteArray(), outInput1.getByteArray());
                                             break;
                                         case LINE:
                                             double startX1 = ADoubleSerializerDeserializer.getDouble(outInput0
-                                                    .getBytes(), ALineSerializerDeserializer
+                                                    .getByteArray(), ALineSerializerDeserializer
                                                     .getStartPointCoordinateOffset(Coordinate.X));
                                             double startY1 = ADoubleSerializerDeserializer.getDouble(outInput0
-                                                    .getBytes(), ALineSerializerDeserializer
+                                                    .getByteArray(), ALineSerializerDeserializer
                                                     .getStartPointCoordinateOffset(Coordinate.Y));
                                             double endX1 = ADoubleSerializerDeserializer.getDouble(
-                                                    outInput0.getBytes(), ALineSerializerDeserializer
+                                                    outInput0.getByteArray(), ALineSerializerDeserializer
                                                             .getEndPointCoordinateOffset(Coordinate.X));
                                             double endY1 = ADoubleSerializerDeserializer.getDouble(
-                                                    outInput0.getBytes(), ALineSerializerDeserializer
+                                                    outInput0.getByteArray(), ALineSerializerDeserializer
                                                             .getEndPointCoordinateOffset(Coordinate.Y));
 
                                             double startX2 = ADoubleSerializerDeserializer.getDouble(outInput1
-                                                    .getBytes(), ALineSerializerDeserializer
+                                                    .getByteArray(), ALineSerializerDeserializer
                                                     .getStartPointCoordinateOffset(Coordinate.X));
                                             double startY2 = ADoubleSerializerDeserializer.getDouble(outInput1
-                                                    .getBytes(), ALineSerializerDeserializer
+                                                    .getByteArray(), ALineSerializerDeserializer
                                                     .getStartPointCoordinateOffset(Coordinate.Y));
                                             double endX2 = ADoubleSerializerDeserializer.getDouble(
-                                                    outInput1.getBytes(), ALineSerializerDeserializer
+                                                    outInput1.getByteArray(), ALineSerializerDeserializer
                                                             .getEndPointCoordinateOffset(Coordinate.X));
                                             double endY2 = ADoubleSerializerDeserializer.getDouble(
-                                                    outInput1.getBytes(), ALineSerializerDeserializer
+                                                    outInput1.getByteArray(), ALineSerializerDeserializer
                                                             .getEndPointCoordinateOffset(Coordinate.Y));
                                             res = lineLineIntersection(startX1, startY1, endX1, endY1, startX2,
                                                     startY2, endX2, endY2);
                                             break;
                                         case POLYGON:
-                                            res = linePolygonIntersection(outInput0.getBytes(), outInput1.getBytes());
+                                            res = linePolygonIntersection(outInput0.getByteArray(), outInput1.getByteArray());
                                             break;
                                         case CIRCLE:
-                                            res = lineCircleIntersection(outInput0.getBytes(), outInput1.getBytes());
+                                            res = lineCircleIntersection(outInput0.getByteArray(), outInput1.getByteArray());
                                             break;
                                         case RECTANGLE:
-                                            res = lineRectangleIntersection(outInput0.getBytes(), outInput1.getBytes());
+                                            res = lineRectangleIntersection(outInput0.getByteArray(), outInput1.getByteArray());
                                             break;
                                         case NULL:
                                             res = false;
@@ -849,26 +849,26 @@
                                 case POLYGON:
                                     switch (tag1) {
                                         case POINT:
-                                            res = pointInPolygon(outInput1.getBytes(), outInput0.getBytes());
+                                            res = pointInPolygon(outInput1.getByteArray(), outInput0.getByteArray());
                                             break;
                                         case LINE:
-                                            res = linePolygonIntersection(outInput1.getBytes(), outInput0.getBytes());
+                                            res = linePolygonIntersection(outInput1.getByteArray(), outInput0.getByteArray());
                                             break;
                                         case POLYGON:
                                             int numOfPoints0 = AInt16SerializerDeserializer.getShort(
-                                                    outInput0.getBytes(),
+                                                    outInput0.getByteArray(),
                                                     APolygonSerializerDeserializer.getNumberOfPointsOffset());
                                             int numOfPoints1 = AInt16SerializerDeserializer.getShort(
-                                                    outInput1.getBytes(),
+                                                    outInput1.getByteArray(),
                                                     APolygonSerializerDeserializer.getNumberOfPointsOffset());
 
                                             if (numOfPoints0 < 3 || numOfPoints1 < 3) {
                                                 throw new AlgebricksException("Polygon must have at least 3 points.");
                                             }
 
-                                            getCounterClockWisePolygon(outInput0.getBytes(), pointsOffsets0,
+                                            getCounterClockWisePolygon(outInput0.getByteArray(), pointsOffsets0,
                                                     numOfPoints0);
-                                            getCounterClockWisePolygon(outInput1.getBytes(), pointsOffsets1,
+                                            getCounterClockWisePolygon(outInput1.getByteArray(), pointsOffsets1,
                                                     numOfPoints1);
                                             int nonSimplePolygonDetection0 = 2 * numOfPoints0;
                                             int nonSimplePolygonDetection1 = 2 * numOfPoints1;
@@ -880,7 +880,7 @@
                                             trianglesX1.reset();
                                             trianglesY1.reset();
                                             while (true) {
-                                                middleVertex1 = triangulatePolygon(outInput1.getBytes(), numOfPoints1,
+                                                middleVertex1 = triangulatePolygon(outInput1.getByteArray(), numOfPoints1,
                                                         pointsOffsets1, trianglesX1, trianglesY1, numOfTriangles1,
                                                         nonSimplePolygonDetection1, middleVertex1);
 
@@ -896,7 +896,7 @@
                                             trianglesX0.reset();
                                             trianglesY0.reset();
                                             while (true) {
-                                                middleVertex0 = triangulatePolygon(outInput0.getBytes(), numOfPoints0,
+                                                middleVertex0 = triangulatePolygon(outInput0.getByteArray(), numOfPoints0,
                                                         pointsOffsets0, trianglesX0, trianglesY0, numOfTriangles0,
                                                         nonSimplePolygonDetection0, middleVertex0);
 
@@ -929,11 +929,11 @@
                                             }
                                             break;
                                         case CIRCLE:
-                                            res = polygonCircleIntersection(outInput0.getBytes(), outInput1.getBytes());
+                                            res = polygonCircleIntersection(outInput0.getByteArray(), outInput1.getByteArray());
                                             break;
                                         case RECTANGLE:
-                                            res = rectanglePolygonIntersection(outInput1.getBytes(),
-                                                    outInput0.getBytes());
+                                            res = rectanglePolygonIntersection(outInput1.getByteArray(),
+                                                    outInput0.getByteArray());
                                             break;
                                         case NULL:
                                             res = false;
@@ -948,20 +948,20 @@
                                 case CIRCLE:
                                     switch (tag1) {
                                         case POINT:
-                                            res = pointInCircle(outInput1.getBytes(), outInput0.getBytes());
+                                            res = pointInCircle(outInput1.getByteArray(), outInput0.getByteArray());
                                             break;
                                         case LINE:
-                                            res = lineCircleIntersection(outInput1.getBytes(), outInput0.getBytes());
+                                            res = lineCircleIntersection(outInput1.getByteArray(), outInput0.getByteArray());
                                             break;
                                         case POLYGON:
-                                            res = polygonCircleIntersection(outInput1.getBytes(), outInput0.getBytes());
+                                            res = polygonCircleIntersection(outInput1.getByteArray(), outInput0.getByteArray());
                                             break;
                                         case CIRCLE:
-                                            res = circleCircleIntersection(outInput0.getBytes(), outInput1.getBytes());
+                                            res = circleCircleIntersection(outInput0.getByteArray(), outInput1.getByteArray());
                                             break;
                                         case RECTANGLE:
-                                            res = rectangleCircleIntersection(outInput1.getBytes(),
-                                                    outInput0.getBytes());
+                                            res = rectangleCircleIntersection(outInput1.getByteArray(),
+                                                    outInput0.getByteArray());
                                             break;
                                         case NULL:
                                             res = false;
@@ -976,22 +976,22 @@
                                 case RECTANGLE:
                                     switch (tag1) {
                                         case POINT:
-                                            res = pointInRectangle(outInput1.getBytes(), outInput0.getBytes());
+                                            res = pointInRectangle(outInput1.getByteArray(), outInput0.getByteArray());
                                             break;
                                         case LINE:
-                                            res = lineRectangleIntersection(outInput1.getBytes(), outInput0.getBytes());
+                                            res = lineRectangleIntersection(outInput1.getByteArray(), outInput0.getByteArray());
                                             break;
                                         case POLYGON:
-                                            res = rectanglePolygonIntersection(outInput0.getBytes(),
-                                                    outInput1.getBytes());
+                                            res = rectanglePolygonIntersection(outInput0.getByteArray(),
+                                                    outInput1.getByteArray());
                                             break;
                                         case CIRCLE:
-                                            res = rectangleCircleIntersection(outInput0.getBytes(),
-                                                    outInput1.getBytes());
+                                            res = rectangleCircleIntersection(outInput0.getByteArray(),
+                                                    outInput1.getByteArray());
                                             break;
                                         case RECTANGLE:
-                                            triangulateRectangle(outInput0.getBytes(), trianglesX0, trianglesY0);
-                                            triangulateRectangle(outInput1.getBytes(), trianglesX1, trianglesY1);
+                                            triangulateRectangle(outInput0.getByteArray(), trianglesX0, trianglesY0);
+                                            triangulateRectangle(outInput1.getByteArray(), trianglesX1, trianglesY1);
 
                                             boolean intersect = false;
                                             // 2 triangles in a rectangle
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SubstringDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SubstringDescriptor.java
index c2678b5..8f3ca50 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SubstringDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SubstringDescriptor.java
@@ -49,14 +49,14 @@
                     public void evaluate(IFrameTupleReference tuple) throws AlgebricksException {
                         argOut.reset();
                         evalStart.evaluate(tuple);
-                        int start = IntegerSerializerDeserializer.getInt(argOut.getBytes(), 1) - 1;
+                        int start = IntegerSerializerDeserializer.getInt(argOut.getByteArray(), 1) - 1;
                         argOut.reset();
                         evalLen.evaluate(tuple);
-                        int len = IntegerSerializerDeserializer.getInt(argOut.getBytes(), 1);
+                        int len = IntegerSerializerDeserializer.getInt(argOut.getByteArray(), 1);
                         argOut.reset();
                         evalString.evaluate(tuple);
 
-                        byte[] bytes = argOut.getBytes();
+                        byte[] bytes = argOut.getByteArray();
                         int utflen = UTF8StringPointable.getUTFLen(bytes, 1);
                         int sStart = 3;
                         int c = 0;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SwitchCaseDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SwitchCaseDescriptor.java
index 1c1ca13..93ad6fb 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SwitchCaseDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/SwitchCaseDescriptor.java
@@ -71,7 +71,7 @@
                                 if (equals(condOut, caseOut)) {
                                     argOut.reset();
                                     evals[i + 1].evaluate(tuple);
-                                    output.getDataOutput().write(argOut.getBytes(), argOut.getStartIndex(),
+                                    output.getDataOutput().write(argOut.getByteArray(), argOut.getStartOffset(),
                                             argOut.getLength());
                                     return;
                                 }
@@ -79,7 +79,7 @@
                             // the default case
                             argOut.reset();
                             evals[n - 1].evaluate(tuple);
-                            output.getDataOutput().write(argOut.getBytes(), argOut.getStartIndex(), argOut.getLength());
+                            output.getDataOutput().write(argOut.getByteArray(), argOut.getStartOffset(), argOut.getLength());
                         } catch (HyracksDataException hde) {
                             throw new AlgebricksException(hde);
                         } catch (IOException ioe) {
@@ -88,11 +88,11 @@
                     }
 
                     private boolean equals(ArrayBackedValueStorage out1, ArrayBackedValueStorage out2) {
-                        if (out1.getStartIndex() != out2.getStartIndex() || out1.getLength() != out2.getLength())
+                        if (out1.getStartOffset() != out2.getStartOffset() || out1.getLength() != out2.getLength())
                             return false;
-                        byte[] data1 = out1.getBytes();
-                        byte[] data2 = out2.getBytes();
-                        for (int i = out1.getStartIndex(); i < out1.getLength(); i++) {
+                        byte[] data1 = out1.getByteArray();
+                        byte[] data2 = out2.getByteArray();
+                        for (int i = out1.getStartOffset(); i < out1.getLength(); i++) {
                             if (data1[i] != data2[i]) {
                                 return false;
                             }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/YearDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/YearDescriptor.java
index 65be0bd..03a5ddc 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/YearDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/evaluators/functions/YearDescriptor.java
@@ -72,7 +72,7 @@
                         try {
                             out1.reset();
                             eval1.evaluate(tuple);
-                            byte[] dateArray = out1.getBytes();
+                            byte[] dateArray = out1.getByteArray();
 
                             if (dateArray[0] == SER_NULL_TYPE_TAG) {
                                 nullSerde.serialize(ANull.NULL, out);
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 0fa62aa..efff2cd 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
@@ -366,7 +366,7 @@
                 } catch (HyracksDataException e) {
                     throw new AlgebricksException(e);
                 }
-                IEvaluatorFactory fldIndexEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs.getBytes(),
+                IEvaluatorFactory fldIndexEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs.getByteArray(),
                         abvs.getLength()));
                 IEvaluatorFactory evalFactory = new FieldAccessByIndexEvalFactory(recordEvalFactory,
                         fldIndexEvalFactory, recType);
@@ -392,7 +392,7 @@
         } catch (HyracksDataException e) {
             throw new AlgebricksException(e);
         }
-        IEvaluatorFactory dimensionEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs1.getBytes(),
+        IEvaluatorFactory dimensionEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs1.getByteArray(),
                 abvs1.getLength()));
 
         for (int i = 0; i < numOfFields; i++) {
@@ -404,7 +404,7 @@
             } catch (HyracksDataException e) {
                 throw new AlgebricksException(e);
             }
-            IEvaluatorFactory coordinateEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs2.getBytes(),
+            IEvaluatorFactory coordinateEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs2.getByteArray(),
                     abvs2.getLength()));
 
             evalFactories[i] = new CreateMBREvalFactory(evalFactory, dimensionEvalFactory, coordinateEvalFactory);
@@ -431,7 +431,7 @@
                 } catch (HyracksDataException e) {
                     throw new AlgebricksException(e);
                 }
-                IEvaluatorFactory fldIndexEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs.getBytes(),
+                IEvaluatorFactory fldIndexEvalFactory = new ConstantEvalFactory(Arrays.copyOf(abvs.getByteArray(),
                         abvs.getLength()));
                 IEvaluatorFactory evalFactory = new FieldAccessByIndexEvalFactory(recordEvalFactory,
                         fldIndexEvalFactory, recType);
@@ -569,7 +569,7 @@
         } catch (HyracksDataException e) {
             throw new AlgebricksException(e);
         }
-        return new ConstantEvalFactory(Arrays.copyOf(abvs.getBytes(), abvs.getLength()));
+        return new ConstantEvalFactory(Arrays.copyOf(abvs.getByteArray(), abvs.getLength()));
     }
 
     @Override
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/operators/file/AdmSchemafullRecordParserFactory.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/operators/file/AdmSchemafullRecordParserFactory.java
index 46d1ba3..43556f2 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/operators/file/AdmSchemafullRecordParserFactory.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/operators/file/AdmSchemafullRecordParserFactory.java
@@ -25,8 +25,8 @@
 import edu.uci.ics.asterix.dataflow.data.nontagged.serde.ADateTimeSerializerDeserializer;
 import edu.uci.ics.asterix.dataflow.data.nontagged.serde.ADurationSerializerDeserializer;
 import edu.uci.ics.asterix.dataflow.data.nontagged.serde.ALineSerializerDeserializer;
-import edu.uci.ics.asterix.dataflow.data.nontagged.serde.APointSerializerDeserializer;
 import edu.uci.ics.asterix.dataflow.data.nontagged.serde.APoint3DSerializerDeserializer;
+import edu.uci.ics.asterix.dataflow.data.nontagged.serde.APointSerializerDeserializer;
 import edu.uci.ics.asterix.dataflow.data.nontagged.serde.APolygonSerializerDeserializer;
 import edu.uci.ics.asterix.dataflow.data.nontagged.serde.ARectangleSerializerDeserializer;
 import edu.uci.ics.asterix.dataflow.data.nontagged.serde.ATimeSerializerDeserializer;
@@ -793,11 +793,11 @@
                             token = nextToken();
                             this.admFromLexerStream(token, fieldType, fieldValueBuffer.getDataOutput(), false);
                             if (openRecordField) {
-                                if (fieldValueBuffer.getBytes()[0] != ATypeTag.NULL.serialize())
+                                if (fieldValueBuffer.getByteArray()[0] != ATypeTag.NULL.serialize())
                                     recBuilder.addField(fieldNameBuffer, fieldValueBuffer);
                             } else if (recType.getFieldTypes()[fieldId].getTypeTag() == ATypeTag.UNION) {
                                 if (NonTaggedFormatUtil.isOptionalField((AUnionType) recType.getFieldTypes()[fieldId])) {
-                                    if (fieldValueBuffer.getBytes()[0] != ATypeTag.NULL.serialize()) {
+                                    if (fieldValueBuffer.getByteArray()[0] != ATypeTag.NULL.serialize()) {
                                         recBuilder.addField(fieldId, fieldValueBuffer);
                                     }
                                 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/operators/file/AdmTupleParser.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/operators/file/AdmTupleParser.java
index 3b52315..573eb78 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/operators/file/AdmTupleParser.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/operators/file/AdmTupleParser.java
@@ -468,14 +468,14 @@
 				this.admFromLexerStream(token, fieldType,
 						fieldValueBuffer.getDataOutput(), false);
 				if (openRecordField) {
-					if (fieldValueBuffer.getBytes()[0] != ATypeTag.NULL
+					if (fieldValueBuffer.getByteArray()[0] != ATypeTag.NULL
 							.serialize())
 						recBuilder.addField(fieldNameBuffer, fieldValueBuffer);
 				} else if (recType.getFieldTypes()[fieldId].getTypeTag() == ATypeTag.UNION) {
 					if (NonTaggedFormatUtil
 							.isOptionalField((AUnionType) recType
 									.getFieldTypes()[fieldId])) {
-						if (fieldValueBuffer.getBytes()[0] != ATypeTag.NULL
+						if (fieldValueBuffer.getByteArray()[0] != ATypeTag.NULL
 								.serialize()) {
 							recBuilder.addField(fieldId, fieldValueBuffer);
 						}
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/AFlatValuePointable.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/AFlatValuePointable.java
new file mode 100644
index 0000000..41d55b6
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/AFlatValuePointable.java
@@ -0,0 +1,81 @@
+/*
+ * 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.pointables;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.runtime.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.runtime.pointables.visitor.IVisitablePointableVisitor;
+import edu.uci.ics.asterix.runtime.util.container.IElementFactory;
+import edu.uci.ics.hyracks.data.std.api.IValueReference;
+
+/**
+ * This class is to represent a flat field, e.g., int field, string field, and
+ * so on, based on an binary representation
+ */
+public class AFlatValuePointable extends AbstractVisitablePointable {
+
+    /**
+     * DO NOT allow to create AFlatValuePointable object arbitrarily, force to
+     * use object pool based allocator. The factory is not public so that it
+     * cannot called in other places than PointableAllocator.
+     */
+    static IElementFactory<IVisitablePointable, IAType> FACTORY = new IElementFactory<IVisitablePointable, IAType>() {
+        public AFlatValuePointable createElement(IAType type) {
+            return new AFlatValuePointable();
+        }
+    };
+
+    /**
+     * private constructor, to prevent arbitrary creation
+     */
+    private AFlatValuePointable() {
+
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof IValueReference))
+            return false;
+
+        // get right raw data
+        IValueReference ivf = (IValueReference) o;
+        byte[] odata = ivf.getByteArray();
+        int ostart = ivf.getStartOffset();
+        int olen = ivf.getLength();
+
+        // get left raw data
+        byte[] data = getByteArray();
+        int start = getStartOffset();
+        int len = getLength();
+
+        // bytes length should be equal
+        if (len != olen)
+            return false;
+
+        // check each byte
+        for (int i = 0; i < len; i++) {
+            if (data[start + i] != odata[ostart + i])
+                return false;
+        }
+        return true;
+    }
+
+    @Override
+    public <R, T> R accept(IVisitablePointableVisitor<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/pointables/AListPointable.java
similarity index 63%
rename from asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/AListAccessor.java
rename to asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/AListPointable.java
index b3a5108..9bcf3f5 100644
--- 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/pointables/AListPointable.java
@@ -13,7 +13,7 @@
  * limitations under the License.
  */
 
-package edu.uci.ics.asterix.runtime.accessors;
+package edu.uci.ics.asterix.runtime.pointables;
 
 import java.io.DataOutputStream;
 import java.util.ArrayList;
@@ -26,33 +26,47 @@
 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.AccessorAllocator;
+import edu.uci.ics.asterix.runtime.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.runtime.pointables.visitor.IVisitablePointableVisitor;
 import edu.uci.ics.asterix.runtime.util.ResettableByteArrayOutputStream;
 import edu.uci.ics.asterix.runtime.util.container.IElementFactory;
 
-public class AListAccessor extends AbstractBinaryAccessor {
+/**
+ * This class is to interpret the binary data representation of a list, one can
+ * call getItems and getItemTags to get pointable objects for items and item
+ * type tags
+ * 
+ */
+public class AListPointable extends AbstractVisitablePointable {
 
-    public static IElementFactory<IBinaryAccessor, IAType> FACTORY = new IElementFactory<IBinaryAccessor, IAType>() {
-        public IBinaryAccessor createElement(IAType type) {
-            return new AListAccessor((AbstractCollectionType) type);
+    /**
+     * DO NOT allow to create AListPointable object arbitrarily, force to use
+     * object pool based allocator, in order to have object reuse
+     */
+    static IElementFactory<IVisitablePointable, IAType> FACTORY = new IElementFactory<IVisitablePointable, IAType>() {
+        public IVisitablePointable createElement(IAType type) {
+            return new AListPointable((AbstractCollectionType) type);
         }
     };
 
+    private final List<IVisitablePointable> items = new ArrayList<IVisitablePointable>();
+    private final List<IVisitablePointable> itemTags = new ArrayList<IVisitablePointable>();
+    private final PointableAllocator allocator = new PointableAllocator();
+
+    private final ResettableByteArrayOutputStream dataBos = new ResettableByteArrayOutputStream();
+    private final DataOutputStream dataDos = new DataOutputStream(dataBos);
+
     private IAType itemType;
     private ATypeTag itemTag;
     private boolean typedItemList = false;
 
-    private List<IBinaryAccessor> items = new ArrayList<IBinaryAccessor>();
-    private List<IBinaryAccessor> itemTags = new ArrayList<IBinaryAccessor>();
-    private AccessorAllocator allocator = new AccessorAllocator();
-
-    private byte[] dataBuffer = new byte[32768];
-    private ResettableByteArrayOutputStream dataBos = new ResettableByteArrayOutputStream();
-    private DataOutputStream dataDos = new DataOutputStream(dataBos);
-
-    private AListAccessor(AbstractCollectionType inputType) {
+    /**
+     * private constructor, to prevent constructing it arbitrarily
+     * 
+     * @param inputType
+     *            , the input type
+     */
+    private AListPointable(AbstractCollectionType inputType) {
         if (inputType != null && inputType.getItemType() != null) {
             itemType = inputType.getItemType();
             if (itemType.getTypeTag() == ATypeTag.ANY) {
@@ -70,11 +84,11 @@
         allocator.reset();
         items.clear();
         itemTags.clear();
-        dataBos.setByteArray(dataBuffer, 0);
+        dataBos.reset();
     }
 
     @Override
-    public void reset(byte[] b, int s, int len) {
+    public void set(byte[] b, int s, int len) {
         reset();
 
         int numberOfitems = AInt32SerializerDeserializer.getInt(b, s + 6);
@@ -91,29 +105,30 @@
                 default:
                     itemOffset = s + 10;
             }
-        } else
+        } else {
             itemOffset = s + 10 + (numberOfitems * 4);
+        }
         int itemLength = 0;
         try {
             if (typedItemList) {
                 for (int i = 0; i < numberOfitems; i++) {
                     itemLength = NonTaggedFormatUtil.getFieldValueLength(b, itemOffset, itemTag, false);
-                    IBinaryAccessor tag = allocator.allocateFieldType();
-                    IBinaryAccessor item = allocator.allocateFieldValue(itemType);
+                    IVisitablePointable tag = allocator.allocateEmpty();
+                    IVisitablePointable item = allocator.allocateFieldValue(itemType);
 
                     // set item type tag
                     int start = dataBos.size();
                     dataDos.writeByte(itemTag.serialize());
                     int end = dataBos.size();
-                    tag.reset(dataBuffer, start, end - start);
+                    tag.set(dataBos.getByteArray(), start, end - start);
                     itemTags.add(tag);
-                    
+
                     // set item value
                     start = dataBos.size();
                     dataDos.writeByte(itemTag.serialize());
                     dataDos.write(b, itemOffset, itemLength);
                     end = dataBos.size();
-                    item.reset(dataBuffer, start, end - start);
+                    item.set(dataBos.getByteArray(), start, end - start);
                     itemOffset += itemLength;
                     items.add(item);
                 }
@@ -121,18 +136,18 @@
                 for (int i = 0; i < numberOfitems; i++) {
                     itemTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b[itemOffset]);
                     itemLength = NonTaggedFormatUtil.getFieldValueLength(b, itemOffset, itemTag, true) + 1;
-                    IBinaryAccessor tag = allocator.allocateFieldType();
-                    IBinaryAccessor item = allocator.allocateFieldValue(itemType);
+                    IVisitablePointable tag = allocator.allocateEmpty();
+                    IVisitablePointable item = allocator.allocateFieldValue(itemType);
 
                     // set item type tag
                     int start = dataBos.size();
                     dataDos.writeByte(itemTag.serialize());
                     int end = dataBos.size();
-                    tag.reset(dataBuffer, start, end - start);
+                    tag.set(dataBos.getByteArray(), start, end - start);
                     itemTags.add(tag);
 
                     // open part field already include the type tag
-                    item.reset(b, itemOffset, itemLength);
+                    item.set(b, itemOffset, itemLength);
                     itemOffset += itemLength;
                     items.add(item);
                 }
@@ -143,15 +158,15 @@
     }
 
     @Override
-    public <R, T> R accept(IBinaryAccessorVisitor<R, T> vistor, T tag) throws AsterixException {
+    public <R, T> R accept(IVisitablePointableVisitor<R, T> vistor, T tag) throws AsterixException {
         return vistor.visit(this, tag);
     }
 
-    public List<IBinaryAccessor> getItems() {
+    public List<IVisitablePointable> getItems() {
         return items;
     }
 
-    public List<IBinaryAccessor> getItemTags() {
+    public List<IVisitablePointable> getItemTags() {
         return itemTags;
     }
 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/ARecordAccessor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/ARecordPointable.java
similarity index 70%
rename from asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/ARecordAccessor.java
rename to asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/ARecordPointable.java
index 55cea26..5bfcb83 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/ARecordAccessor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/ARecordPointable.java
@@ -13,7 +13,7 @@
  * limitations under the License.
  */
 
-package edu.uci.ics.asterix.runtime.accessors;
+package edu.uci.ics.asterix.runtime.pointables;
 
 import java.io.DataOutputStream;
 import java.io.IOException;
@@ -29,47 +29,61 @@
 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.AccessorAllocator;
+import edu.uci.ics.asterix.runtime.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.runtime.pointables.visitor.IVisitablePointableVisitor;
 import edu.uci.ics.asterix.runtime.util.ResettableByteArrayOutputStream;
 import edu.uci.ics.asterix.runtime.util.container.IElementFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.INullWriter;
 
-public class ARecordAccessor extends AbstractBinaryAccessor {
+/**
+ * This class is to interpret the binary data representation of a record, one
+ * can call getFieldNames, getFieldTypeTags and getFieldValues to get pointable
+ * objects for field names, field type tags, and field values.
+ * 
+ */
+public class ARecordPointable extends AbstractVisitablePointable {
 
-    public static IElementFactory<IBinaryAccessor, IAType> FACTORY = new IElementFactory<IBinaryAccessor, IAType>() {
-        public IBinaryAccessor createElement(IAType type) {
-            return new ARecordAccessor((ARecordType) type);
+    /**
+     * DO NOT allow to create ARecordPointable object arbitrarily, force to use
+     * object pool based allocator, in order to have object reuse
+     */
+    static IElementFactory<IVisitablePointable, IAType> FACTORY = new IElementFactory<IVisitablePointable, IAType>() {
+        public IVisitablePointable createElement(IAType type) {
+            return new ARecordPointable((ARecordType) type);
         }
     };
 
     // access results: field names, field types, and field values
-    private List<IBinaryAccessor> fieldNames = new ArrayList<IBinaryAccessor>();
-    private List<IBinaryAccessor> fieldTypeTags = new ArrayList<IBinaryAccessor>();
-    private List<IBinaryAccessor> fieldValues = new ArrayList<IBinaryAccessor>();
+    private final List<IVisitablePointable> fieldNames = new ArrayList<IVisitablePointable>();
+    private final List<IVisitablePointable> fieldTypeTags = new ArrayList<IVisitablePointable>();
+    private final List<IVisitablePointable> fieldValues = new ArrayList<IVisitablePointable>();
 
-    // accessor allocator
-    AccessorAllocator allocator = new AccessorAllocator();
+    // pointable allocator
+    private final PointableAllocator allocator = new PointableAllocator();
 
-    private byte[] typeBuffer = new byte[32768];
-    private ResettableByteArrayOutputStream typeBos = new ResettableByteArrayOutputStream();
-    private DataOutputStream typeDos = new DataOutputStream(typeBos);
+    private final ResettableByteArrayOutputStream typeBos = new ResettableByteArrayOutputStream();
+    private final DataOutputStream typeDos = new DataOutputStream(typeBos);
 
-    private byte[] dataBuffer = new byte[32768];
-    private ResettableByteArrayOutputStream dataBos = new ResettableByteArrayOutputStream();
-    private DataOutputStream dataDos = new DataOutputStream(dataBos);
+    private final ResettableByteArrayOutputStream dataBos = new ResettableByteArrayOutputStream();
+    private final DataOutputStream dataDos = new DataOutputStream(dataBos);
+
+    private final ARecordType inputRecType;
+
+    private final int numberOfSchemaFields;
+    private final int[] fieldOffsets;
+    private final IVisitablePointable nullReference = AFlatValuePointable.FACTORY.createElement(null);
 
     private int closedPartTypeInfoSize = 0;
-    private ARecordType inputRecType;
-
-    private int numberOfSchemaFields;
     private int offsetArrayOffset;
-    private int[] fieldOffsets;
     private ATypeTag typeTag;
-    private IBinaryAccessor nullReference = AFlatValueAccessor.FACTORY.createElement(null);
 
-    public ARecordAccessor(ARecordType inputType) {
+    /**
+     * private constructor, to prevent constructing it arbitrarily
+     * 
+     * @param inputType
+     *            , the input type
+     */
+    private ARecordPointable(ARecordType inputType) {
         this.inputRecType = inputType;
         IAType[] fieldTypes = inputType.getFieldTypes();
         String[] fieldNameStrs = inputType.getFieldNames();
@@ -77,7 +91,7 @@
 
         // initialize the buffer for closed parts(fieldName bytes+ type bytes) +
         // constant(null bytes)
-        typeBos.setByteArray(typeBuffer, 0);
+        typeBos.reset();
         try {
             for (int i = 0; i < numberOfSchemaFields; i++) {
                 ATypeTag ftypeTag = fieldTypes[i].getTypeTag();
@@ -92,8 +106,8 @@
                 int tagStart = typeBos.size();
                 typeDos.writeByte(ftypeTag.serialize());
                 int tagEnd = typeBos.size();
-                IBinaryAccessor typeTagReference = AFlatValueAccessor.FACTORY.createElement(null);
-                typeTagReference.reset(typeBuffer, tagStart, tagEnd - tagStart);
+                IVisitablePointable typeTagReference = AFlatValuePointable.FACTORY.createElement(null);
+                typeTagReference.set(typeBos.getByteArray(), tagStart, tagEnd - tagStart);
                 fieldTypeTags.add(typeTagReference);
 
                 // add type name Reference (including a astring type tag)
@@ -101,8 +115,8 @@
                 typeDos.writeByte(ATypeTag.STRING.serialize());
                 typeDos.writeUTF(fieldNameStrs[i]);
                 int nameEnd = typeBos.size();
-                IBinaryAccessor typeNameReference = AFlatValueAccessor.FACTORY.createElement(null);
-                typeNameReference.reset(typeBuffer, nameStart, nameEnd - nameStart);
+                IVisitablePointable typeNameReference = AFlatValuePointable.FACTORY.createElement(null);
+                typeNameReference.set(typeBos.getByteArray(), nameStart, nameEnd - nameStart);
                 fieldNames.add(typeNameReference);
             }
 
@@ -111,7 +125,7 @@
             INullWriter nullWriter = AqlNullWriterFactory.INSTANCE.createNullWriter();
             nullWriter.writeNull(typeDos);
             int nullFieldEnd = typeBos.size();
-            nullReference.reset(typeBuffer, nullFieldStart, nullFieldEnd - nullFieldStart);
+            nullReference.set(typeBos.getByteArray(), nullFieldStart, nullFieldEnd - nullFieldStart);
         } catch (IOException e) {
             throw new IllegalStateException(e);
         }
@@ -120,8 +134,8 @@
     }
 
     private void reset() {
-        typeBos.setByteArray(typeBuffer, closedPartTypeInfoSize);
-        dataBos.setByteArray(dataBuffer, 0);
+        typeBos.reset(closedPartTypeInfoSize);
+        dataBos.reset(0);
         // reset the allocator
         allocator.reset();
 
@@ -133,10 +147,11 @@
         fieldValues.clear();
     }
 
-    public void reset(byte[] b, int start, int len) {
+    @Override
+    public void set(byte[] b, int start, int len) {
         // clear the previous states
         reset();
-        super.reset(b, start, len);
+        super.set(b, start, len);
 
         boolean isExpanded = false;
         int openPartOffset = 0;
@@ -152,10 +167,12 @@
                 if (isExpanded) {
                     openPartOffset = s + AInt32SerializerDeserializer.getInt(b, s + 6);
                     s += 10;
-                } else
+                } else {
                     s += 6;
-            } else
+                }
+            } else {
                 s += 5;
+            }
         }
         try {
             if (numberOfSchemaFields > 0) {
@@ -206,8 +223,8 @@
                     dataDos.writeByte(typeTag.serialize());
                     dataDos.write(b, fieldOffsets[fieldNumber], fieldValueLength);
                     int fend = dataBos.size();
-                    IBinaryAccessor fieldValue = allocator.allocateFieldValue(fieldType);
-                    fieldValue.reset(dataBuffer, fstart, fend - fstart);
+                    IVisitablePointable fieldValue = allocator.allocateFieldValue(fieldType);
+                    fieldValue.set(dataBos.getByteArray(), fstart, fend - fstart);
                     fieldValues.add(fieldValue);
                 }
             }
@@ -223,14 +240,14 @@
                     dataDos.writeByte(ATypeTag.STRING.serialize());
                     dataDos.write(b, fieldOffset, fieldValueLength);
                     int fnend = dataBos.size();
-                    IBinaryAccessor fieldName = allocator.allocateFieldName();
-                    fieldName.reset(dataBuffer, fnstart, fnend - fnstart);
+                    IVisitablePointable fieldName = allocator.allocateEmpty();
+                    fieldName.set(dataBos.getByteArray(), fnstart, fnend - fnstart);
                     fieldNames.add(fieldName);
                     fieldOffset += fieldValueLength;
 
                     // set the field type tag
-                    IBinaryAccessor fieldTypeTag = allocator.allocateFieldType();
-                    fieldTypeTag.reset(b, fieldOffset, 1);
+                    IVisitablePointable fieldTypeTag = allocator.allocateEmpty();
+                    fieldTypeTag.set(b, fieldOffset, 1);
                     fieldTypeTags.add(fieldTypeTag);
                     typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b[fieldOffset]);
 
@@ -238,8 +255,8 @@
                     fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b, fieldOffset, typeTag, true) + 1;
 
                     // allocate
-                    IBinaryAccessor fieldValueAccessor = allocator.allocateFieldValue(typeTag);
-                    fieldValueAccessor.reset(b, fieldOffset, fieldValueLength);
+                    IVisitablePointable fieldValueAccessor = allocator.allocateFieldValue(typeTag);
+                    fieldValueAccessor.set(b, fieldOffset, fieldValueLength);
                     fieldValues.add(fieldValueAccessor);
                     fieldOffset += fieldValueLength;
                 }
@@ -249,20 +266,20 @@
         }
     }
 
-    public List<IBinaryAccessor> getFieldNames() {
+    public List<IVisitablePointable> getFieldNames() {
         return fieldNames;
     }
 
-    public List<IBinaryAccessor> getFieldTypeTags() {
+    public List<IVisitablePointable> getFieldTypeTags() {
         return fieldTypeTags;
     }
 
-    public List<IBinaryAccessor> getFieldValues() {
+    public List<IVisitablePointable> getFieldValues() {
         return fieldValues;
     }
 
     @Override
-    public <R, T> R accept(IBinaryAccessorVisitor<R, T> vistor, T tag) throws AsterixException {
+    public <R, T> R accept(IVisitablePointableVisitor<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/pointables/AbstractVisitablePointable.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/AbstractVisitablePointable.java
new file mode 100644
index 0000000..6b844a0
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/AbstractVisitablePointable.java
@@ -0,0 +1,59 @@
+/*
+ * 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.pointables;
+
+import edu.uci.ics.asterix.runtime.pointables.base.IVisitablePointable;
+import edu.uci.ics.hyracks.data.std.api.IValueReference;
+
+/**
+ * This class implements several "routine" methods in IVisitablePointable
+ * interface, so that subclasses do not need to repeat the same code
+ * 
+ */
+public abstract class AbstractVisitablePointable implements IVisitablePointable {
+
+    private byte[] data;
+    private int start;
+    private int len;
+
+    @Override
+    public byte[] getByteArray() {
+        return data;
+    }
+
+    @Override
+    public int getLength() {
+        return len;
+    }
+
+    @Override
+    public int getStartOffset() {
+        return start;
+    }
+
+    @Override
+    public void set(byte[] b, int start, int len) {
+        this.data = b;
+        this.start = start;
+        this.len = len;
+    }
+
+    @Override
+    public void set(IValueReference ivf) {
+        set(ivf.getByteArray(), ivf.getStartOffset(), ivf.getLength());
+    }
+
+}
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/PointableAllocator.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/PointableAllocator.java
new file mode 100644
index 0000000..7675a2b
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/PointableAllocator.java
@@ -0,0 +1,92 @@
+/*
+ * 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.pointables;
+
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.runtime.pointables.base.DefaultOpenFieldType;
+import edu.uci.ics.asterix.runtime.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.runtime.util.container.IElementAllocator;
+import edu.uci.ics.asterix.runtime.util.container.ListElementAllocator;
+
+/**
+ * This class is the ONLY place to create IVisitablePointable object instances,
+ * so that all the client code are forced to have object reuse for those
+ * IVisitablePointable objects
+ */
+public class PointableAllocator {
+
+    private IElementAllocator<IVisitablePointable, IAType> flatValueAllocator = new ListElementAllocator<IVisitablePointable, IAType>(
+            AFlatValuePointable.FACTORY);
+    private IElementAllocator<IVisitablePointable, IAType> recordValueAllocator = new ListElementAllocator<IVisitablePointable, IAType>(
+            ARecordPointable.FACTORY);
+    private IElementAllocator<IVisitablePointable, IAType> listValueAllocator = new ListElementAllocator<IVisitablePointable, IAType>(
+            AListPointable.FACTORY);
+
+    public IVisitablePointable allocateEmpty() {
+        return flatValueAllocator.allocate(null);
+    }
+
+    /**
+     * allocate closed part value pointable
+     * 
+     * @param type
+     * @return the pointable object
+     */
+    public IVisitablePointable allocateFieldValue(IAType type) {
+        if (type == null)
+            return flatValueAllocator.allocate(null);
+        else if (type.getTypeTag().equals(ATypeTag.RECORD))
+            return recordValueAllocator.allocate(type);
+        else if (type.getTypeTag().equals(ATypeTag.UNORDEREDLIST) || type.getTypeTag().equals(ATypeTag.ORDEREDLIST))
+            return listValueAllocator.allocate(type);
+        else
+            return flatValueAllocator.allocate(null);
+    }
+
+    /**
+     * allocate open part value pointable
+     * 
+     * @param typeTag
+     * @return the pointable object
+     */
+    public IVisitablePointable allocateFieldValue(ATypeTag typeTag) {
+        if (typeTag == null)
+            return flatValueAllocator.allocate(null);
+        else if (typeTag.equals(ATypeTag.RECORD))
+            return recordValueAllocator.allocate(DefaultOpenFieldType.NESTED_OPEN_RECORD_TYPE);
+        else if (typeTag.equals(ATypeTag.UNORDEREDLIST))
+            return listValueAllocator.allocate(DefaultOpenFieldType.NESTED_OPEN_AUNORDERED_LIST_TYPE);
+        else if (typeTag.equals(ATypeTag.ORDEREDLIST))
+            return listValueAllocator.allocate(DefaultOpenFieldType.NESTED_OPEN_AORDERED_LIST_TYPE);
+        else
+            return flatValueAllocator.allocate(null);
+    }
+
+    public IVisitablePointable allocateListValue(IAType type) {
+        return listValueAllocator.allocate(type);
+    }
+
+    public IVisitablePointable allocateRecordValue(IAType type) {
+        return recordValueAllocator.allocate(type);
+    }
+
+    public void reset() {
+        flatValueAllocator.reset();
+        recordValueAllocator.reset();
+        listValueAllocator.reset();
+    }
+}
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/base/DefaultOpenFieldType.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/base/DefaultOpenFieldType.java
similarity index 76%
rename from asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/base/DefaultOpenFieldType.java
rename to asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/base/DefaultOpenFieldType.java
index 9da464f..0c39aa0 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/base/DefaultOpenFieldType.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/base/DefaultOpenFieldType.java
@@ -13,7 +13,7 @@
  * limitations under the License.
  */
 
-package edu.uci.ics.asterix.runtime.accessors.base;
+package edu.uci.ics.asterix.runtime.pointables.base;
 
 import edu.uci.ics.asterix.om.types.AOrderedListType;
 import edu.uci.ics.asterix.om.types.ARecordType;
@@ -22,6 +22,13 @@
 import edu.uci.ics.asterix.om.types.BuiltinType;
 import edu.uci.ics.asterix.om.types.IAType;
 
+/**
+ * This class serves as the repository for default record type and list type
+ * fields in the open part, e.g., a "record" (nested) field in the open part is
+ * always a fully open one, and a "list" field in the open part is always a list
+ * of "ANY"
+ * 
+ */
 public class DefaultOpenFieldType {
 
     // nested open field rec type
@@ -35,13 +42,13 @@
     // nested open list type
     public static AUnorderedListType NESTED_OPEN_AUNORDERED_LIST_TYPE = new AUnorderedListType(BuiltinType.ANY,
             "nested-unordered-list");
-    
-    public static IAType getDefaultOpenFieldType(ATypeTag tag){
-        if(tag.equals(ATypeTag.RECORD))
+
+    public static IAType getDefaultOpenFieldType(ATypeTag tag) {
+        if (tag.equals(ATypeTag.RECORD))
             return NESTED_OPEN_RECORD_TYPE;
-        if(tag.equals(ATypeTag.ORDEREDLIST))
+        if (tag.equals(ATypeTag.ORDEREDLIST))
             return NESTED_OPEN_AORDERED_LIST_TYPE;
-        if(tag.equals(ATypeTag.UNORDEREDLIST))
+        if (tag.equals(ATypeTag.UNORDEREDLIST))
             return NESTED_OPEN_AUNORDERED_LIST_TYPE;
         else
             return null;
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/base/IVisitablePointable.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/base/IVisitablePointable.java
new file mode 100644
index 0000000..f630e24
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/base/IVisitablePointable.java
@@ -0,0 +1,29 @@
+/*
+ * 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.pointables.base;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.runtime.pointables.visitor.IVisitablePointableVisitor;
+import edu.uci.ics.hyracks.data.std.api.IPointable;
+
+/**
+ * This interface extends IPointable with a visitor interface in order to ease
+ * programming for recursive record structures
+ */
+public interface IVisitablePointable extends IPointable {
+
+    public <R, T> R accept(IVisitablePointableVisitor<R, T> vistor, T tag) throws AsterixException;
+}
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/ACastVisitor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/ACastVisitor.java
new file mode 100644
index 0000000..9d3fe73
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/ACastVisitor.java
@@ -0,0 +1,89 @@
+/*
+ * 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.pointables.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.AbstractCollectionType;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.runtime.pointables.AFlatValuePointable;
+import edu.uci.ics.asterix.runtime.pointables.AListPointable;
+import edu.uci.ics.asterix.runtime.pointables.ARecordPointable;
+import edu.uci.ics.asterix.runtime.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.runtime.pointables.visitor.IVisitablePointableVisitor;
+import edu.uci.ics.hyracks.algebricks.common.utils.Triple;
+
+/**
+ * This class is a IVisitablePointableVisitor implementation which recursively
+ * visit a given record, list or flat value of a given type, and recursively
+ * cast it to a specified type For example,
+ * 
+ * A record { "hobby": {{"music", "coding"}}, "id": "001", "name":
+ * "Person Three"} which confirms to closed type ( id: string, name: string,
+ * hobby: {{string}}? ) can be casted to a open type (id: string )
+ * 
+ * Since the open/close part of a record has completely different underlying
+ * memory/storage layout, the visitor will change the layout as specified at
+ * runtime.
+ */
+public class ACastVisitor implements IVisitablePointableVisitor<Void, Triple<IVisitablePointable, IAType, Boolean>> {
+
+    private final Map<IVisitablePointable, ARecordCaster> raccessorToCaster = new HashMap<IVisitablePointable, ARecordCaster>();
+    private final Map<IVisitablePointable, AListCaster> laccessorToCaster = new HashMap<IVisitablePointable, AListCaster>();
+
+    @Override
+    public Void visit(AListPointable accessor, Triple<IVisitablePointable, IAType, Boolean> arg)
+            throws AsterixException {
+        AListCaster caster = laccessorToCaster.get(accessor);
+        if (caster == null) {
+            caster = new AListCaster();
+            laccessorToCaster.put(accessor, caster);
+        }
+        try {
+            caster.castList(accessor, arg.first, (AbstractCollectionType) arg.second, this);
+        } catch (Exception e) {
+            throw new AsterixException(e);
+        }
+        return null;
+    }
+
+    @Override
+    public Void visit(ARecordPointable accessor, Triple<IVisitablePointable, 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(AFlatValuePointable accessor, Triple<IVisitablePointable, IAType, Boolean> arg) {
+        // set the pointer for result
+        arg.first.set(accessor);
+        return null;
+    }
+
+}
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/cast/AListCaster.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/AListCaster.java
similarity index 62%
rename from asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/cast/AListCaster.java
rename to asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/AListCaster.java
index aba06b5..d3e4a12 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/cast/AListCaster.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/AListCaster.java
@@ -13,7 +13,7 @@
  * limitations under the License.
  */
 
-package edu.uci.ics.asterix.runtime.accessors.cast;
+package edu.uci.ics.asterix.runtime.pointables.cast;
 
 import java.io.DataOutput;
 import java.io.DataOutputStream;
@@ -29,50 +29,57 @@
 import edu.uci.ics.asterix.om.types.AbstractCollectionType;
 import edu.uci.ics.asterix.om.types.EnumDeserializer;
 import edu.uci.ics.asterix.om.types.IAType;
-import edu.uci.ics.asterix.runtime.accessors.AFlatValueAccessor;
-import edu.uci.ics.asterix.runtime.accessors.AListAccessor;
-import edu.uci.ics.asterix.runtime.accessors.base.DefaultOpenFieldType;
-import edu.uci.ics.asterix.runtime.accessors.base.IBinaryAccessor;
+import edu.uci.ics.asterix.runtime.pointables.AListPointable;
+import edu.uci.ics.asterix.runtime.pointables.PointableAllocator;
+import edu.uci.ics.asterix.runtime.pointables.base.DefaultOpenFieldType;
+import edu.uci.ics.asterix.runtime.pointables.base.IVisitablePointable;
 import edu.uci.ics.asterix.runtime.util.ResettableByteArrayOutputStream;
 import edu.uci.ics.hyracks.algebricks.common.utils.Triple;
 
+/**
+ * This class is to do the runtime type cast for a list. It is ONLY visible to
+ * ACastVisitor, so that no other client places can call into that caster and
+ * unnecessary bugs could be prevented.
+ */
 class AListCaster {
+    // pointable allocator
+    private final PointableAllocator allocator = new PointableAllocator();
 
-    private IAType reqItemType;
-    private IBinaryAccessor itemTempReference = AFlatValueAccessor.FACTORY.createElement(null);
-    private Triple<IBinaryAccessor, IAType, Boolean> itemVisitorArg = new Triple<IBinaryAccessor, IAType, Boolean>(
+    // for storing the cast result
+    private final IVisitablePointable itemTempReference = allocator.allocateEmpty();
+    private final Triple<IVisitablePointable, IAType, Boolean> itemVisitorArg = new Triple<IVisitablePointable, IAType, Boolean>(
             itemTempReference, null, null);
 
-    private UnorderedListBuilder unOrderedListBuilder = new UnorderedListBuilder();
-    private OrderedListBuilder orderedListBuilder = new OrderedListBuilder();
+    private final UnorderedListBuilder unOrderedListBuilder = new UnorderedListBuilder();
+    private final OrderedListBuilder orderedListBuilder = new OrderedListBuilder();
 
-    private byte[] dataBuffer = new byte[32768];
-    private ResettableByteArrayOutputStream dataBos = new ResettableByteArrayOutputStream();
-    private DataOutput dataDos = new DataOutputStream(dataBos);
+    private final ResettableByteArrayOutputStream dataBos = new ResettableByteArrayOutputStream();
+    private final DataOutput dataDos = new DataOutputStream(dataBos);
+    private IAType reqItemType;
 
     public AListCaster() {
 
     }
 
-    public void castList(AListAccessor listAccessor, IBinaryAccessor resultAccessor, AbstractCollectionType reqType,
-            ACastVisitor visitor) throws IOException, AsterixException {
+    public void castList(AListPointable listAccessor, IVisitablePointable resultAccessor,
+            AbstractCollectionType reqType, ACastVisitor visitor) throws IOException, AsterixException {
         if (reqType.getTypeTag().equals(ATypeTag.UNORDEREDLIST)) {
             unOrderedListBuilder.reset((AUnorderedListType) reqType);
         }
         if (reqType.getTypeTag().equals(ATypeTag.ORDEREDLIST)) {
             orderedListBuilder.reset((AOrderedListType) reqType);
         }
-        dataBos.setByteArray(dataBuffer, 0);
+        dataBos.reset();
 
-        List<IBinaryAccessor> itemTags = listAccessor.getItemTags();
-        List<IBinaryAccessor> items = listAccessor.getItems();
+        List<IVisitablePointable> itemTags = listAccessor.getItemTags();
+        List<IVisitablePointable> items = listAccessor.getItems();
 
         int start = dataBos.size();
         for (int i = 0; i < items.size(); i++) {
-            IBinaryAccessor itemTypeTag = itemTags.get(i);
-            IBinaryAccessor item = items.get(i);
-            ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(itemTypeTag.getBytes()[itemTypeTag
-                    .getStartIndex()]);
+            IVisitablePointable itemTypeTag = itemTags.get(i);
+            IVisitablePointable item = items.get(i);
+            ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(itemTypeTag.getByteArray()[itemTypeTag
+                    .getStartOffset()]);
             if (reqItemType == null || reqItemType.getTypeTag().equals(ATypeTag.ANY)) {
                 itemVisitorArg.second = DefaultOpenFieldType.getDefaultOpenFieldType(typeTag);
                 item.accept(visitor, itemVisitorArg);
@@ -96,6 +103,6 @@
             unOrderedListBuilder.write(dataDos, true);
         }
         int end = dataBos.size();
-        resultAccessor.reset(dataBuffer, start, end - start);
+        resultAccessor.set(dataBos.getByteArray(), start, end - start);
     }
 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/cast/ARecordCaster.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/ARecordCaster.java
similarity index 70%
rename from asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/cast/ARecordCaster.java
rename to asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/ARecordCaster.java
index a4efa58..561c113 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/accessors/cast/ARecordCaster.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/ARecordCaster.java
@@ -13,7 +13,7 @@
  * limitations under the License.
  */
 
-package edu.uci.ics.asterix.runtime.accessors.cast;
+package edu.uci.ics.asterix.runtime.pointables.cast;
 
 import java.io.DataOutput;
 import java.io.DataOutputStream;
@@ -30,20 +30,52 @@
 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.AFlatValueAccessor;
-import edu.uci.ics.asterix.runtime.accessors.ARecordAccessor;
-import edu.uci.ics.asterix.runtime.accessors.base.DefaultOpenFieldType;
-import edu.uci.ics.asterix.runtime.accessors.base.IBinaryAccessor;
+import edu.uci.ics.asterix.runtime.pointables.ARecordPointable;
+import edu.uci.ics.asterix.runtime.pointables.PointableAllocator;
+import edu.uci.ics.asterix.runtime.pointables.base.DefaultOpenFieldType;
+import edu.uci.ics.asterix.runtime.pointables.base.IVisitablePointable;
 import edu.uci.ics.asterix.runtime.util.ResettableByteArrayOutputStream;
 import edu.uci.ics.hyracks.algebricks.common.utils.Triple;
 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.api.IValueReference;
 import edu.uci.ics.hyracks.data.std.primitive.UTF8StringPointable;
-import edu.uci.ics.hyracks.dataflow.common.data.accessors.IValueReference;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ByteArrayAccessibleOutputStream;
 
+/**
+ * This class is to do the runtime type cast for a record. It is ONLY visible to
+ * ACastVisitor, so that no other client places can call into that caster and
+ * unnecessary bugs could be prevented.
+ */
 class ARecordCaster {
 
+    // pointable allocator
+    private final PointableAllocator allocator = new PointableAllocator();
+
+    private final List<IVisitablePointable> reqFieldNames = new ArrayList<IVisitablePointable>();
+    private final List<IVisitablePointable> reqFieldTypeTags = new ArrayList<IVisitablePointable>();
+    private ARecordType cachedReqType = null;
+
+    private final ResettableByteArrayOutputStream bos = new ResettableByteArrayOutputStream();
+    private final DataOutputStream dos = new DataOutputStream(bos);
+
+    private final RecordBuilder recBuilder = new RecordBuilder();
+    private final IVisitablePointable nullReference = allocator.allocateEmpty();
+    private final IVisitablePointable nullTypeTag = allocator.allocateEmpty();
+
+    private final IBinaryComparator fieldNameComparator = PointableBinaryComparatorFactory.of(
+            UTF8StringPointable.FACTORY).createBinaryComparator();
+
+    private final ByteArrayAccessibleOutputStream outputBos = new ByteArrayAccessibleOutputStream();
+    private final DataOutputStream outputDos = new DataOutputStream(outputBos);
+
+    private final IVisitablePointable fieldTempReference = allocator.allocateEmpty();
+    private final Triple<IVisitablePointable, IAType, Boolean> nestedVisitorArg = new Triple<IVisitablePointable, IAType, Boolean>(
+            fieldTempReference, null, null);
+
+    private int numInputFields = 0;
+
     // describe closed fields in the required type
     private int[] fieldPermutation;
     private boolean[] optionalFields;
@@ -51,54 +83,30 @@
     // describe fields (open or not) in the input records
     private boolean[] openFields;
     private int[] fieldNamesSortedIndex;
-
-    private List<IBinaryAccessor> reqFieldNames = new ArrayList<IBinaryAccessor>();
     private int[] reqFieldNamesSortedIndex;
-    private List<IBinaryAccessor> reqFieldTypeTags = new ArrayList<IBinaryAccessor>();
-    private ARecordType cachedReqType = null;
-
-    private byte[] buffer = new byte[32768];
-    private ResettableByteArrayOutputStream bos = new ResettableByteArrayOutputStream();
-    private DataOutputStream dos = new DataOutputStream(bos);
-
-    private RecordBuilder recBuilder = new RecordBuilder();
-    private IBinaryAccessor nullReference = AFlatValueAccessor.FACTORY.createElement(null);
-    private IBinaryAccessor nullTypeTag = AFlatValueAccessor.FACTORY.createElement(null);
-
-    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);
-
-    private IBinaryAccessor fieldTempReference = AFlatValueAccessor.FACTORY.createElement(null);
-    private Triple<IBinaryAccessor, IAType, Boolean> nestedVisitorArg = new Triple<IBinaryAccessor, IAType, Boolean>(
-            fieldTempReference, null, null);
 
     public ARecordCaster() {
         try {
-            bos.setByteArray(buffer, 0);
+            bos.reset();
             int start = bos.size();
             INullWriter nullWriter = AqlNullWriterFactory.INSTANCE.createNullWriter();
             nullWriter.writeNull(dos);
             int end = bos.size();
-            nullReference.reset(buffer, start, end - start);
+            nullReference.set(bos.getByteArray(), start, end - start);
             start = bos.size();
             dos.write(ATypeTag.NULL.serialize());
             end = bos.size();
-            nullTypeTag.reset(buffer, start, end);
+            nullTypeTag.set(bos.getByteArray(), start, end);
         } catch (IOException e) {
             throw new IllegalStateException(e);
         }
     }
 
-    public void castRecord(ARecordAccessor recordAccessor, IBinaryAccessor resultAccessor, ARecordType reqType,
+    public void castRecord(ARecordPointable recordAccessor, IVisitablePointable resultAccessor, ARecordType reqType,
             ACastVisitor visitor) throws IOException, AsterixException {
-        List<IBinaryAccessor> fieldNames = recordAccessor.getFieldNames();
-        List<IBinaryAccessor> fieldTypeTags = recordAccessor.getFieldTypeTags();
-        List<IBinaryAccessor> fieldValues = recordAccessor.getFieldValues();
+        List<IVisitablePointable> fieldNames = recordAccessor.getFieldNames();
+        List<IVisitablePointable> fieldTypeTags = recordAccessor.getFieldTypeTags();
+        List<IVisitablePointable> fieldValues = recordAccessor.getFieldValues();
         numInputFields = fieldNames.size();
 
         if (openFields == null || numInputFields > openFields.length) {
@@ -113,7 +121,7 @@
         reset();
         matchClosedPart(fieldNames, fieldTypeTags, fieldValues);
         writeOutput(fieldNames, fieldTypeTags, fieldValues, outputDos, visitor);
-        resultAccessor.reset(outputBuffer, 0, outputBos.size());
+        resultAccessor.set(outputBos.getByteArray(), 0, outputBos.size());
     }
 
     private void reset() {
@@ -123,7 +131,7 @@
             fieldPermutation[i] = -1;
         for (int i = 0; i < numInputFields; i++)
             fieldNamesSortedIndex[i] = i;
-        outputBos.setByteArray(outputBuffer, 0);
+        outputBos.reset();
     }
 
     private void loadRequiredType(ARecordType reqType) throws IOException {
@@ -139,7 +147,7 @@
         for (int i = 0; i < optionalFields.length; i++)
             optionalFields[i] = false;
 
-        bos.setByteArray(buffer, nullReference.getStartIndex() + nullReference.getLength());
+        bos.reset(nullReference.getStartOffset() + nullReference.getLength());
         for (int i = 0; i < numSchemaFields; i++) {
             ATypeTag ftypeTag = fieldTypes[i].getTypeTag();
             String fname = fieldNames[i];
@@ -155,8 +163,8 @@
             int tagStart = bos.size();
             dos.writeByte(ftypeTag.serialize());
             int tagEnd = bos.size();
-            IBinaryAccessor typeTagPointable = AFlatValueAccessor.FACTORY.createElement(null);
-            typeTagPointable.reset(buffer, tagStart, tagEnd - tagStart);
+            IVisitablePointable typeTagPointable = allocator.allocateEmpty();
+            typeTagPointable.set(bos.getByteArray(), tagStart, tagEnd - tagStart);
             reqFieldTypeTags.add(typeTagPointable);
 
             // add type name pointable (including a string type tag)
@@ -164,8 +172,8 @@
             dos.write(ATypeTag.STRING.serialize());
             dos.writeUTF(fname);
             int nameEnd = bos.size();
-            IBinaryAccessor typeNamePointable = AFlatValueAccessor.FACTORY.createElement(null);
-            typeNamePointable.reset(buffer, nameStart, nameEnd - nameStart);
+            IVisitablePointable typeNamePointable = allocator.allocateEmpty();
+            typeNamePointable.set(bos.getByteArray(), nameStart, nameEnd - nameStart);
             reqFieldNames.add(typeNamePointable);
         }
 
@@ -176,8 +184,8 @@
         quickSort(reqFieldNamesSortedIndex, reqFieldNames, 0, reqFieldNamesSortedIndex.length - 1);
     }
 
-    private void matchClosedPart(List<IBinaryAccessor> fieldNames, List<IBinaryAccessor> fieldTypeTags,
-            List<IBinaryAccessor> fieldValues) {
+    private void matchClosedPart(List<IVisitablePointable> fieldNames, List<IVisitablePointable> fieldTypeTags,
+            List<IVisitablePointable> fieldValues) {
         // sort-merge based match
         quickSort(fieldNamesSortedIndex, fieldNames, 0, numInputFields - 1);
         int fnStart = 0;
@@ -187,8 +195,8 @@
             int reqFnPos = reqFieldNamesSortedIndex[reqFnStart];
             int c = compare(fieldNames.get(fnPos), reqFieldNames.get(reqFnPos));
             if (c == 0) {
-                IBinaryAccessor fieldTypeTag = fieldTypeTags.get(fnPos);
-                IBinaryAccessor reqFieldTypeTag = reqFieldTypeTags.get(reqFnPos);
+                IVisitablePointable fieldTypeTag = fieldTypeTags.get(fnPos);
+                IVisitablePointable reqFieldTypeTag = reqFieldTypeTags.get(reqFnPos);
                 if (fieldTypeTag.equals(reqFieldTypeTag) || (
                 // match the null type of optional field
                         optionalFields[reqFnPos] && fieldTypeTag.equals(nullTypeTag))) {
@@ -222,8 +230,8 @@
         }
     }
 
-    private void writeOutput(List<IBinaryAccessor> fieldNames, List<IBinaryAccessor> fieldTypeTags,
-            List<IBinaryAccessor> fieldValues, DataOutput output, ACastVisitor visitor) throws IOException,
+    private void writeOutput(List<IVisitablePointable> fieldNames, List<IVisitablePointable> fieldTypeTags,
+            List<IVisitablePointable> fieldValues, DataOutput output, ACastVisitor visitor) throws IOException,
             AsterixException {
         // reset the states of the record builder
         recBuilder.reset(cachedReqType);
@@ -232,7 +240,7 @@
         // write the closed part
         for (int i = 0; i < fieldPermutation.length; i++) {
             int pos = fieldPermutation[i];
-            IBinaryAccessor field;
+            IVisitablePointable field;
             if (pos >= 0) {
                 field = fieldValues.get(pos);
             } else {
@@ -254,12 +262,12 @@
         // write the open part
         for (int i = 0; i < numInputFields; i++) {
             if (openFields[i]) {
-                IBinaryAccessor name = fieldNames.get(i);
-                IBinaryAccessor field = fieldValues.get(i);
-                IBinaryAccessor fieldTypeTag = fieldTypeTags.get(i);
+                IVisitablePointable name = fieldNames.get(i);
+                IVisitablePointable field = fieldValues.get(i);
+                IVisitablePointable fieldTypeTag = fieldTypeTags.get(i);
 
                 ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
-                        .deserialize(fieldTypeTag.getBytes()[fieldTypeTag.getStartIndex()]);
+                        .deserialize(fieldTypeTag.getByteArray()[fieldTypeTag.getStartOffset()]);
                 nestedVisitorArg.second = DefaultOpenFieldType.getDefaultOpenFieldType(typeTag);
                 field.accept(visitor, nestedVisitorArg);
                 recBuilder.addField(name, nestedVisitorArg.first);
@@ -268,7 +276,7 @@
         recBuilder.write(output, true);
     }
 
-    private void quickSort(int[] index, List<IBinaryAccessor> names, int start, int end) {
+    private void quickSort(int[] index, List<IVisitablePointable> names, int start, int end) {
         if (end <= start)
             return;
         int i = partition(index, names, start, end);
@@ -276,7 +284,7 @@
         quickSort(index, names, i + 1, end);
     }
 
-    private int partition(int[] index, List<IBinaryAccessor> names, int left, int right) {
+    private int partition(int[] index, List<IVisitablePointable> names, int left, int right) {
         int i = left - 1;
         int j = right;
         while (true) {
@@ -305,7 +313,7 @@
 
     private int compare(IValueReference a, IValueReference b) {
         // start+1 and len-1 due to the type tag
-        return fieldNameComparator.compare(a.getBytes(), a.getStartIndex() + 1, a.getLength() - 1, b.getBytes(),
-                b.getStartIndex() + 1, b.getLength() - 1);
+        return fieldNameComparator.compare(a.getByteArray(), a.getStartOffset() + 1, a.getLength() - 1,
+                b.getByteArray(), b.getStartOffset() + 1, b.getLength() - 1);
     }
 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/visitor/IVisitablePointableVisitor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/visitor/IVisitablePointableVisitor.java
new file mode 100644
index 0000000..3f99509
--- /dev/null
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/visitor/IVisitablePointableVisitor.java
@@ -0,0 +1,30 @@
+/*
+ * 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.pointables.visitor;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.runtime.pointables.AFlatValuePointable;
+import edu.uci.ics.asterix.runtime.pointables.AListPointable;
+import edu.uci.ics.asterix.runtime.pointables.ARecordPointable;
+
+public interface IVisitablePointableVisitor<R, T> {
+
+    public R visit(AListPointable accessor, T arg) throws AsterixException;
+
+    public R visit(ARecordPointable accessor, T arg) throws AsterixException;
+
+    public R visit(AFlatValuePointable accessor, T arg) throws AsterixException;
+}
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/unnestingfunctions/std/RangeDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/unnestingfunctions/std/RangeDescriptor.java
index d3b965f..9ca7596 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/unnestingfunctions/std/RangeDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/unnestingfunctions/std/RangeDescriptor.java
@@ -65,10 +65,10 @@
                     public void init(IFrameTupleReference tuple) throws AlgebricksException {
                         inputVal.reset();
                         eval0.evaluate(tuple);
-                        current = IntegerSerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                        current = IntegerSerializerDeserializer.getInt(inputVal.getByteArray(), 1);
                         inputVal.reset();
                         eval1.evaluate(tuple);
-                        max = IntegerSerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                        max = IntegerSerializerDeserializer.getInt(inputVal.getByteArray(), 1);
                     }
 
                     @SuppressWarnings("unchecked")
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/unnestingfunctions/std/ScanCollectionDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/unnestingfunctions/std/ScanCollectionDescriptor.java
index 6bf5420..d59d6c4 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/unnestingfunctions/std/ScanCollectionDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/unnestingfunctions/std/ScanCollectionDescriptor.java
@@ -87,7 +87,7 @@
                     try {
                         inputVal.reset();
                         argEval.evaluate(tuple);
-                        byte[] serList = inputVal.getBytes();
+                        byte[] serList = inputVal.getByteArray();
 
                         if (serList[0] == SER_NULL_TYPE_TAG) {
                             nullSerde.serialize(ANull.NULL, out);
@@ -99,12 +99,12 @@
                                     + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(serList[0]));
                         }
 
-                        serListTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(inputVal.getBytes()[0])
+                        serListTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(inputVal.getByteArray()[0])
                                 .serialize();
                         if (serListTag == SER_ORDEREDLIST_TYPE_TAG)
-                            numItems = AOrderedListSerializerDeserializer.getNumberOfItems(inputVal.getBytes());
+                            numItems = AOrderedListSerializerDeserializer.getNumberOfItems(inputVal.getByteArray());
                         else
-                            numItems = AUnorderedListSerializerDeserializer.getNumberOfItems(inputVal.getBytes());
+                            numItems = AUnorderedListSerializerDeserializer.getNumberOfItems(inputVal.getByteArray());
 
                         itemTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(serList[1]);
                         if (itemTag == ATypeTag.ANY)
@@ -123,7 +123,7 @@
 
                     try {
                         if (pos < numItems) {
-                            byte[] serList = inputVal.getBytes();
+                            byte[] serList = inputVal.getByteArray();
 
                             try {
                                 if (serListTag == SER_ORDEREDLIST_TYPE_TAG) {
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/unnestingfunctions/std/SubsetCollectionDescriptor.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/unnestingfunctions/std/SubsetCollectionDescriptor.java
index a045344..6214d27 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/unnestingfunctions/std/SubsetCollectionDescriptor.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/unnestingfunctions/std/SubsetCollectionDescriptor.java
@@ -75,16 +75,16 @@
                         try {
                             inputVal.reset();
                             evalStart.evaluate(tuple);
-                            posStart = IntegerSerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                            posStart = IntegerSerializerDeserializer.getInt(inputVal.getByteArray(), 1);
 
                             inputVal.reset();
                             evalLen.evaluate(tuple);
-                            numItems = IntegerSerializerDeserializer.getInt(inputVal.getBytes(), 1);
+                            numItems = IntegerSerializerDeserializer.getInt(inputVal.getByteArray(), 1);
 
                             inputVal.reset();
                             evalList.evaluate(tuple);
 
-                            byte[] serList = inputVal.getBytes();
+                            byte[] serList = inputVal.getByteArray();
 
                             if (serList[0] == SER_NULL_TYPE_TAG) {
                                 nullSerde.serialize(ANull.NULL, out);
@@ -113,7 +113,7 @@
                     @Override
                     public boolean step() throws AlgebricksException {
                         if (posCrt < posStart + numItems && posCrt < numItemsMax) {
-                            byte[] serList = inputVal.getBytes();
+                            byte[] serList = inputVal.getByteArray();
                             int itemLength = 0;
                             try {
                                 int itemOffset = AOrderedListSerializerDeserializer.getItemOffset(serList, posCrt);
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/AccessorAllocator.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/AccessorAllocator.java
deleted file mode 100644
index dde5ff6..0000000
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/AccessorAllocator.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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 edu.uci.ics.asterix.om.types.ATypeTag;
-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.base.DefaultOpenFieldType;
-import edu.uci.ics.asterix.runtime.accessors.base.IBinaryAccessor;
-import edu.uci.ics.asterix.runtime.util.container.IElementAllocator;
-import edu.uci.ics.asterix.runtime.util.container.ListElementAllocator;
-
-public class AccessorAllocator {
-
-    private IElementAllocator<IBinaryAccessor, IAType> flatArtifactAllocator = new ListElementAllocator<IBinaryAccessor, IAType>(
-            AFlatValueAccessor.FACTORY);
-    private IElementAllocator<IBinaryAccessor, IAType> nestedRecValueAllocator = new ListElementAllocator<IBinaryAccessor, IAType>(
-            ARecordAccessor.FACTORY);
-    private IElementAllocator<IBinaryAccessor, IAType> nestedListValueAllocator = new ListElementAllocator<IBinaryAccessor, IAType>(
-            AListAccessor.FACTORY);
-
-    public IBinaryAccessor allocateFieldName() {
-        return flatArtifactAllocator.allocate(null);
-    }
-
-    public IBinaryAccessor allocateFieldType() {
-        return flatArtifactAllocator.allocate(null);
-    }
-
-    public IBinaryAccessor allocateFieldValue(IAType type) {
-        if(type == null)
-            return flatArtifactAllocator.allocate(null);
-        else if (type.getTypeTag().equals(ATypeTag.RECORD))
-            return nestedRecValueAllocator.allocate(type);
-        else if (type.getTypeTag().equals(ATypeTag.UNORDEREDLIST) || type.getTypeTag().equals(ATypeTag.ORDEREDLIST))
-            return nestedListValueAllocator.allocate(type);
-        else
-            return flatArtifactAllocator.allocate(null);
-    }
-    
-    public IBinaryAccessor allocateFieldValue(ATypeTag typeTag) {
-        if(typeTag == null)
-            return flatArtifactAllocator.allocate(null);
-        else if (typeTag.equals(ATypeTag.RECORD))
-            return nestedRecValueAllocator.allocate(DefaultOpenFieldType.NESTED_OPEN_RECORD_TYPE);
-        else if (typeTag.equals(ATypeTag.UNORDEREDLIST))
-            return nestedListValueAllocator.allocate(DefaultOpenFieldType.NESTED_OPEN_AUNORDERED_LIST_TYPE);
-        else if(typeTag.equals(ATypeTag.ORDEREDLIST))
-            return nestedListValueAllocator.allocate(DefaultOpenFieldType.NESTED_OPEN_AORDERED_LIST_TYPE);
-        else
-            return flatArtifactAllocator.allocate(null);
-    }
-
-    public IBinaryAccessor allocateNestedListValue(IAType type) {
-        return nestedListValueAllocator.allocate(type);
-    }
-
-    public void reset() {
-        flatArtifactAllocator.reset();
-        nestedRecValueAllocator.reset();
-        nestedListValueAllocator.reset();
-    }
-}
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ResettableByteArrayInputStream.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ResettableByteArrayInputStream.java
deleted file mode 100644
index af18133..0000000
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ResettableByteArrayInputStream.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * 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 ResettableByteArrayInputStream extends ByteArrayInputStream {
-    private static final Logger LOGGER = Logger.getLogger(ResettableByteArrayInputStream.class.getName());
-
-    private byte[] data;
-    private int position;
-
-    public ResettableByteArrayInputStream(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/ResettableByteArrayOutputStream.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ResettableByteArrayOutputStream.java
index 62db3a9..117b7f4 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ResettableByteArrayOutputStream.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/ResettableByteArrayOutputStream.java
@@ -1,63 +1,10 @@
-/*
- * 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;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ByteArrayAccessibleOutputStream;
 
-public class ResettableByteArrayOutputStream extends ByteArrayOutputStream {
-    private static final Logger LOGGER = Logger.getLogger(ResettableByteArrayOutputStream.class.getName());
+public class ResettableByteArrayOutputStream extends ByteArrayAccessibleOutputStream {
 
-    private byte[] data;
-    private int position;
-
-    public ResettableByteArrayOutputStream() {
+    public void reset(int size) {
+        count = size;
     }
-
-    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/container/IElementAllocator.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/container/IElementAllocator.java
index 8d17cb7..8172e56 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/container/IElementAllocator.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/container/IElementAllocator.java
@@ -15,9 +15,24 @@
 
 package edu.uci.ics.asterix.runtime.util.container;
 
+/**
+ * A reusable object pool interface
+ */
 public interface IElementAllocator<E, T> {
 
+    /**
+     * Give client an E instance
+     * 
+     * @param arg
+     *            , the argument to create E
+     * @return a E instance
+     */
     public E allocate(T arg);
 
+    /**
+     * Clean all the used(assigned) instances in the pool. Then all instances in
+     * the pool are marked as "unused" and can be returned to next bunch of
+     * allocate call from a client
+     */
     public void reset();
 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/container/IElementFactory.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/container/IElementFactory.java
index 4698c25..f914d13 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/container/IElementFactory.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/container/IElementFactory.java
@@ -15,7 +15,18 @@
 
 package edu.uci.ics.asterix.runtime.util.container;
 
+/**
+ * A factory interface to create elements. Elements in a reusable pool/container
+ * can ONLY be instantiated from the corresponding factory
+ */
 public interface IElementFactory<E, T> {
 
+    /**
+     * create an element of type E
+     * 
+     * @param arg
+     *            of type T
+     * @return an E element
+     */
     public E createElement(T arg);
 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/container/ListElementAllocator.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/container/ListElementAllocator.java
index 447b740..f54c7c8 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/container/ListElementAllocator.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/container/ListElementAllocator.java
@@ -18,28 +18,92 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.hadoop.io.BooleanWritable;
+
+/**
+ * ListElementAllocator<E, T> is an element-reusable list or a element pool in
+ * other words, however elements in the list should be exactly the same class,
+ * this is forced by IElementFactory<E, T> factory as a parameter to the
+ * constructor once a ListElementAllocator is constructed, it can only store
+ * objects of the same class.
+ * 
+ * The argument for creating E instances could be different. This class also
+ * considers arguments in object reusing, e.g., reuse an E instances ONLY when
+ * the construction argument is "equal".
+ */
 public class ListElementAllocator<E, T> implements IElementAllocator<E, T> {
 
     private IElementFactory<E, T> factory;
+
+    /**
+     * element reusable object pool
+     */
     private List<E> pool = new ArrayList<E>();
-    private int cursor = -1;
+
+    /**
+     * args that are used to create each element in the pool
+     */
+    private List<T> args = new ArrayList<T>();
+
+    /**
+     * bits indicating which element is in use
+     */
+    private List<BooleanWritable> usedBits = new ArrayList<BooleanWritable>();
+
+    /**
+     * the start index for searching
+     */
+    private int minStartIndex = 0;
 
     public ListElementAllocator(IElementFactory<E, T> factory) {
         this.factory = factory;
     }
 
+    @Override
     public E allocate(T arg) {
-        cursor++;
-        if (cursor < pool.size()) {
-            return pool.get(cursor);
-        } else {
-            E element = factory.createElement(arg);
-            pool.add(element);
-            return element;
+        boolean continuous = true;
+        for (int i = minStartIndex; i < pool.size(); i++) {
+            if (!usedBits.get(i).get()) {
+                boolean match = false;
+
+                // the two cases where an element in the pool is a match
+                if ((arg == null && args.get(i) == null)
+                        || (arg != null && args.get(i) != null && arg.equals(args.get(i))))
+                    match = true;
+
+                if (match) {
+                    // the element is not used and the arg is the same as input
+                    // arg
+                    if (continuous) {
+                        minStartIndex++;
+                    }
+                    usedBits.get(i).set(true);
+                    return pool.get(i);
+                } else {
+                    // a unmatched element blocked
+                    // free slots are not continuous from the beginning free
+                    // slot
+                    continuous = false;
+                }
+            } else {
+                if (continuous) {
+                    minStartIndex++;
+                }
+            }
         }
+
+        // if not find a reusable object, allocate a new element
+        E element = factory.createElement(arg);
+        pool.add(element);
+        args.add(arg);
+        usedBits.add(new BooleanWritable(true));
+        return element;
     }
 
+    @Override
     public void reset() {
-        cursor = -1;
+        for (int i = 0; i < usedBits.size(); i++)
+            usedBits.get(i).set(false);
+        minStartIndex = 0;
     }
 }