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 2768902..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
@@ -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 {
 
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/AListPointable.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/AListPointable.java
index 85b5297..9bcf3f5 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/AListPointable.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/AListPointable.java
@@ -53,7 +53,6 @@
     private final List<IVisitablePointable> itemTags = new ArrayList<IVisitablePointable>();
     private final PointableAllocator allocator = new PointableAllocator();
 
-    private final byte[] dataBuffer = new byte[32768];
     private final ResettableByteArrayOutputStream dataBos = new ResettableByteArrayOutputStream();
     private final DataOutputStream dataDos = new DataOutputStream(dataBos);
 
@@ -85,7 +84,7 @@
         allocator.reset();
         items.clear();
         itemTags.clear();
-        dataBos.setByteArray(dataBuffer, 0);
+        dataBos.reset();
     }
 
     @Override
@@ -106,8 +105,9 @@
                 default:
                     itemOffset = s + 10;
             }
-        } else
+        } else {
             itemOffset = s + 10 + (numberOfitems * 4);
+        }
         int itemLength = 0;
         try {
             if (typedItemList) {
@@ -120,7 +120,7 @@
                     int start = dataBos.size();
                     dataDos.writeByte(itemTag.serialize());
                     int end = dataBos.size();
-                    tag.set(dataBuffer, start, end - start);
+                    tag.set(dataBos.getByteArray(), start, end - start);
                     itemTags.add(tag);
 
                     // set item value
@@ -128,7 +128,7 @@
                     dataDos.writeByte(itemTag.serialize());
                     dataDos.write(b, itemOffset, itemLength);
                     end = dataBos.size();
-                    item.set(dataBuffer, start, end - start);
+                    item.set(dataBos.getByteArray(), start, end - start);
                     itemOffset += itemLength;
                     items.add(item);
                 }
@@ -143,7 +143,7 @@
                     int start = dataBos.size();
                     dataDos.writeByte(itemTag.serialize());
                     int end = dataBos.size();
-                    tag.set(dataBuffer, start, end - start);
+                    tag.set(dataBos.getByteArray(), start, end - start);
                     itemTags.add(tag);
 
                     // open part field already include the type tag
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/ARecordPointable.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/ARecordPointable.java
index 8535db3..5bfcb83 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/ARecordPointable.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/ARecordPointable.java
@@ -38,7 +38,7 @@
 /**
  * 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
+ * objects for field names, field type tags, and field values.
  * 
  */
 public class ARecordPointable extends AbstractVisitablePointable {
@@ -61,11 +61,9 @@
     // pointable allocator
     private final PointableAllocator allocator = new PointableAllocator();
 
-    private final byte[] typeBuffer = new byte[32768];
     private final ResettableByteArrayOutputStream typeBos = new ResettableByteArrayOutputStream();
     private final DataOutputStream typeDos = new DataOutputStream(typeBos);
 
-    private final byte[] dataBuffer = new byte[32768];
     private final ResettableByteArrayOutputStream dataBos = new ResettableByteArrayOutputStream();
     private final DataOutputStream dataDos = new DataOutputStream(dataBos);
 
@@ -93,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();
@@ -109,7 +107,7 @@
                 typeDos.writeByte(ftypeTag.serialize());
                 int tagEnd = typeBos.size();
                 IVisitablePointable typeTagReference = AFlatValuePointable.FACTORY.createElement(null);
-                typeTagReference.set(typeBuffer, tagStart, tagEnd - tagStart);
+                typeTagReference.set(typeBos.getByteArray(), tagStart, tagEnd - tagStart);
                 fieldTypeTags.add(typeTagReference);
 
                 // add type name Reference (including a astring type tag)
@@ -118,7 +116,7 @@
                 typeDos.writeUTF(fieldNameStrs[i]);
                 int nameEnd = typeBos.size();
                 IVisitablePointable typeNameReference = AFlatValuePointable.FACTORY.createElement(null);
-                typeNameReference.set(typeBuffer, nameStart, nameEnd - nameStart);
+                typeNameReference.set(typeBos.getByteArray(), nameStart, nameEnd - nameStart);
                 fieldNames.add(typeNameReference);
             }
 
@@ -127,7 +125,7 @@
             INullWriter nullWriter = AqlNullWriterFactory.INSTANCE.createNullWriter();
             nullWriter.writeNull(typeDos);
             int nullFieldEnd = typeBos.size();
-            nullReference.set(typeBuffer, nullFieldStart, nullFieldEnd - nullFieldStart);
+            nullReference.set(typeBos.getByteArray(), nullFieldStart, nullFieldEnd - nullFieldStart);
         } catch (IOException e) {
             throw new IllegalStateException(e);
         }
@@ -136,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();
 
@@ -169,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) {
@@ -224,7 +224,7 @@
                     dataDos.write(b, fieldOffsets[fieldNumber], fieldValueLength);
                     int fend = dataBos.size();
                     IVisitablePointable fieldValue = allocator.allocateFieldValue(fieldType);
-                    fieldValue.set(dataBuffer, fstart, fend - fstart);
+                    fieldValue.set(dataBos.getByteArray(), fstart, fend - fstart);
                     fieldValues.add(fieldValue);
                 }
             }
@@ -241,7 +241,7 @@
                     dataDos.write(b, fieldOffset, fieldValueLength);
                     int fnend = dataBos.size();
                     IVisitablePointable fieldName = allocator.allocateEmpty();
-                    fieldName.set(dataBuffer, fnstart, fnend - fnstart);
+                    fieldName.set(dataBos.getByteArray(), fnstart, fnend - fnstart);
                     fieldNames.add(fieldName);
                     fieldOffset += fieldValueLength;
 
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/AListCaster.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/AListCaster.java
index 9423a0c..d3e4a12 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/AListCaster.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/AListCaster.java
@@ -55,7 +55,6 @@
 
     private final ResettableByteArrayOutputStream dataBos = new ResettableByteArrayOutputStream();
     private final DataOutput dataDos = new DataOutputStream(dataBos);
-    private final byte[] dataBuffer = new byte[32768];
     private IAType reqItemType;
 
     public AListCaster() {
@@ -70,7 +69,7 @@
         if (reqType.getTypeTag().equals(ATypeTag.ORDEREDLIST)) {
             orderedListBuilder.reset((AOrderedListType) reqType);
         }
-        dataBos.setByteArray(dataBuffer, 0);
+        dataBos.reset();
 
         List<IVisitablePointable> itemTags = listAccessor.getItemTags();
         List<IVisitablePointable> items = listAccessor.getItems();
@@ -104,6 +103,6 @@
             unOrderedListBuilder.write(dataDos, true);
         }
         int end = dataBos.size();
-        resultAccessor.set(dataBuffer, start, end - start);
+        resultAccessor.set(dataBos.getByteArray(), start, end - start);
     }
 }
diff --git a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/ARecordCaster.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/ARecordCaster.java
index 70fbc4a..561c113 100644
--- a/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/ARecordCaster.java
+++ b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/pointables/cast/ARecordCaster.java
@@ -57,7 +57,6 @@
     private final List<IVisitablePointable> reqFieldTypeTags = new ArrayList<IVisitablePointable>();
     private ARecordType cachedReqType = null;
 
-    private final byte[] buffer = new byte[32768];
     private final ResettableByteArrayOutputStream bos = new ResettableByteArrayOutputStream();
     private final DataOutputStream dos = new DataOutputStream(bos);
 
@@ -88,16 +87,16 @@
 
     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.set(buffer, start, end - start);
+            nullReference.set(bos.getByteArray(), start, end - start);
             start = bos.size();
             dos.write(ATypeTag.NULL.serialize());
             end = bos.size();
-            nullTypeTag.set(buffer, start, end);
+            nullTypeTag.set(bos.getByteArray(), start, end);
         } catch (IOException e) {
             throw new IllegalStateException(e);
         }
@@ -148,7 +147,7 @@
         for (int i = 0; i < optionalFields.length; i++)
             optionalFields[i] = false;
 
-        bos.setByteArray(buffer, nullReference.getStartOffset() + nullReference.getLength());
+        bos.reset(nullReference.getStartOffset() + nullReference.getLength());
         for (int i = 0; i < numSchemaFields; i++) {
             ATypeTag ftypeTag = fieldTypes[i].getTypeTag();
             String fname = fieldNames[i];
@@ -165,7 +164,7 @@
             dos.writeByte(ftypeTag.serialize());
             int tagEnd = bos.size();
             IVisitablePointable typeTagPointable = allocator.allocateEmpty();
-            typeTagPointable.set(buffer, tagStart, tagEnd - tagStart);
+            typeTagPointable.set(bos.getByteArray(), tagStart, tagEnd - tagStart);
             reqFieldTypeTags.add(typeTagPointable);
 
             // add type name pointable (including a string type tag)
@@ -174,7 +173,7 @@
             dos.writeUTF(fname);
             int nameEnd = bos.size();
             IVisitablePointable typeNamePointable = allocator.allocateEmpty();
-            typeNamePointable.set(buffer, nameStart, nameEnd - nameStart);
+            typeNamePointable.set(bos.getByteArray(), nameStart, nameEnd - nameStart);
             reqFieldNames.add(typeNamePointable);
         }
 
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/ListElementAllocator.java b/asterix-runtime/src/main/java/edu/uci/ics/asterix/runtime/util/container/ListElementAllocator.java
index dc96153..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
@@ -74,9 +74,9 @@
                 if (match) {
                     // the element is not used and the arg is the same as input
                     // arg
-                    if (continuous)
+                    if (continuous) {
                         minStartIndex++;
-
+                    }
                     usedBits.get(i).set(true);
                     return pool.get(i);
                 } else {
@@ -86,8 +86,9 @@
                     continuous = false;
                 }
             } else {
-                if (continuous)
+                if (continuous) {
                     minStartIndex++;
+                }
             }
         }