diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ABooleanPrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ABooleanPrinter.java
index 60ee1fe..33866df 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ABooleanPrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ABooleanPrinter.java
@@ -8,7 +8,6 @@
 
 public class ABooleanPrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final ABooleanPrinter INSTANCE = new ABooleanPrinter();
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ACirclePrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ACirclePrinter.java
index 812d93b..e2da96e 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ACirclePrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ACirclePrinter.java
@@ -8,7 +8,6 @@
 
 public class ACirclePrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final ACirclePrinter INSTANCE = new ACirclePrinter();
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADatePrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADatePrinter.java
index 24aa144..918e286 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADatePrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADatePrinter.java
@@ -9,11 +9,10 @@
 
 public class ADatePrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     private static long CHRONON_OF_DAY = 24 * 60 * 60 * 1000;
     public static final ADatePrinter INSTANCE = new ADatePrinter();
     private static final GregorianCalendarSystem gCalInstance = GregorianCalendarSystem.getInstance();
-    
+
     @Override
     public void init() {
 
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADateTimePrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADateTimePrinter.java
index d856a20..96df767 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADateTimePrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADateTimePrinter.java
@@ -9,7 +9,6 @@
 
 public class ADateTimePrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final ADateTimePrinter INSTANCE = new ADateTimePrinter();
     private static final GregorianCalendarSystem gCalInstance = GregorianCalendarSystem.getInstance();
 
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADoublePrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADoublePrinter.java
index 3a255c6..ae50d0e 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADoublePrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADoublePrinter.java
@@ -8,7 +8,6 @@
 
 public class ADoublePrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final ADoublePrinter INSTANCE = new ADoublePrinter();
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADurationPrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADurationPrinter.java
index a59b9ac..061d372 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADurationPrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ADurationPrinter.java
@@ -12,7 +12,6 @@
 
 public class ADurationPrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final ADurationPrinter INSTANCE = new ADurationPrinter();
     private static final GregorianCalendarSystem gCalInstance = GregorianCalendarSystem.getInstance();
 
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AFloatPrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AFloatPrinter.java
index 27d2e2c..96c9ee9 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AFloatPrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AFloatPrinter.java
@@ -8,7 +8,6 @@
 
 public class AFloatPrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final AFloatPrinter INSTANCE = new AFloatPrinter();
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt16Printer.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt16Printer.java
index 8961013..1687fb1 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt16Printer.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt16Printer.java
@@ -13,8 +13,6 @@
 
 public class AInt16Printer implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
-
     private static final String SUFFIX_STRING = "i16";
     private static byte[] _suffix;
     private static int _suffix_count;
@@ -23,6 +21,7 @@
         DataOutput dout = new DataOutputStream(interm);
         try {
             dout.writeUTF(SUFFIX_STRING);
+            interm.close();
         } catch (IOException e) {
             e.printStackTrace();
         }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt32Printer.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt32Printer.java
index a5c3245..b20db34 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt32Printer.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt32Printer.java
@@ -10,7 +10,6 @@
 
 public class AInt32Printer implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final AInt32Printer INSTANCE = new AInt32Printer();
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt64Printer.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt64Printer.java
index 318828e..312802a 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt64Printer.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt64Printer.java
@@ -13,8 +13,6 @@
 
 public class AInt64Printer implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
-
     private static final String SUFFIX_STRING = "i64";
     private static byte[] _suffix;
     private static int _suffix_count;
@@ -23,6 +21,7 @@
         DataOutput dout = new DataOutputStream(interm);
         try {
             dout.writeUTF(SUFFIX_STRING);
+            interm.close();
         } catch (IOException e) {
             e.printStackTrace();
         }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt8Printer.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt8Printer.java
index bd4139d..a5a421d 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt8Printer.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AInt8Printer.java
@@ -13,8 +13,6 @@
 
 public class AInt8Printer implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
-
     private static final String SUFFIX_STRING = "i8";
     private static byte[] _suffix;
     private static int _suffix_count;
@@ -23,6 +21,7 @@
         DataOutput dout = new DataOutputStream(interm);
         try {
             dout.writeUTF(SUFFIX_STRING);
+            interm.close();
         } catch (IOException e) {
             e.printStackTrace();
         }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ALinePrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ALinePrinter.java
index f99727a..40f405c 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ALinePrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ALinePrinter.java
@@ -8,7 +8,6 @@
 
 public class ALinePrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final ALinePrinter INSTANCE = new ALinePrinter();
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ANullPrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ANullPrinter.java
index 6abc0e1..daea967 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ANullPrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ANullPrinter.java
@@ -5,11 +5,8 @@
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
 import edu.uci.ics.hyracks.algebricks.data.IPrinter;
 
-
-
 public class ANullPrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final ANullPrinter INSTANCE = new ANullPrinter();
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AObjectPrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AObjectPrinter.java
index aff30f7..edcfc8a 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AObjectPrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AObjectPrinter.java
@@ -10,7 +10,6 @@
 
 public class AObjectPrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final AObjectPrinter INSTANCE = new AObjectPrinter();
 
     private IPrinter recordPrinter = new ARecordPrinterFactory(null).createPrinter();
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AOrderedlistPrinterFactory.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AOrderedlistPrinterFactory.java
index 563b14c..7d04ebc 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AOrderedlistPrinterFactory.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AOrderedlistPrinterFactory.java
@@ -2,22 +2,21 @@
 
 import java.io.PrintStream;
 
-import edu.uci.ics.asterix.common.exceptions.AsterixException;
-import edu.uci.ics.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
-import edu.uci.ics.asterix.formats.nontagged.AqlPrinterFactoryProvider;
+import edu.uci.ics.asterix.om.pointables.PointableAllocator;
+import edu.uci.ics.asterix.om.pointables.base.DefaultOpenFieldType;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.pointables.printer.APrintVisitor;
 import edu.uci.ics.asterix.om.types.AOrderedListType;
 import edu.uci.ics.asterix.om.types.ATypeTag;
-import edu.uci.ics.asterix.om.types.EnumDeserializer;
 import edu.uci.ics.asterix.om.types.IAType;
-import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
 import edu.uci.ics.hyracks.algebricks.data.IPrinter;
 import edu.uci.ics.hyracks.algebricks.data.IPrinterFactory;
 
 public class AOrderedlistPrinterFactory implements IPrinterFactory {
 
     private static final long serialVersionUID = 1L;
-
     private AOrderedListType orderedlistType;
 
     public AOrderedlistPrinterFactory(AOrderedListType orderedlistType) {
@@ -27,77 +26,29 @@
     @Override
     public IPrinter createPrinter() {
 
-        return new IPrinter() {
+        PointableAllocator allocator = new PointableAllocator();
+        final IAType inputType = orderedlistType == null ? DefaultOpenFieldType
+                .getDefaultOpenFieldType(ATypeTag.ORDEREDLIST) : orderedlistType;
+        final IVisitablePointable listAccessor = allocator.allocateListValue(inputType);
+        final APrintVisitor printVisitor = new APrintVisitor();
+        final Pair<PrintStream, ATypeTag> arg = new Pair<PrintStream, ATypeTag>(null, null);
 
-            private IPrinter itemPrinter;
-            private IAType itemType;
-            private ATypeTag itemTag;
-            private boolean typedItemList = false;
+        return new IPrinter() {
 
             @Override
             public void init() throws AlgebricksException {
-
-                if (orderedlistType != null && orderedlistType.getItemType() != null) {
-                    itemType = orderedlistType.getItemType();
-                    if (itemType.getTypeTag() == ATypeTag.ANY) {
-                        this.typedItemList = false;
-                        this.itemPrinter = AObjectPrinterFactory.INSTANCE.createPrinter();
-                    } else {
-                        this.typedItemList = true;
-                        itemPrinter = AqlPrinterFactoryProvider.INSTANCE.getPrinterFactory(itemType).createPrinter();
-                        itemTag = orderedlistType.getItemType().getTypeTag();
-                    }
-                } else {
-                    this.typedItemList = false;
-                    this.itemPrinter = AObjectPrinterFactory.INSTANCE.createPrinter();
-                }
-                itemPrinter.init();
-
+                arg.second = inputType.getTypeTag();
             }
 
             @Override
-            public void print(byte[] b, int s, int l, PrintStream ps) throws AlgebricksException {
-                ps.print("[ ");
-                int numberOfitems = AInt32SerializerDeserializer.getInt(b, s + 6);
-                int itemOffset;
-                if (typedItemList) {
-                    switch (itemTag) {
-                        case STRING:
-                        case RECORD:
-                        case ORDEREDLIST:
-                        case UNORDEREDLIST:
-                        case ANY:
-                            itemOffset = s + 10 + (numberOfitems * 4);
-                            break;
-                        default:
-                            itemOffset = s + 10;
-                    }
-                } else
-                    itemOffset = s + 10 + (numberOfitems * 4);
-                int itemLength = 0;
+            public void print(byte[] b, int start, int l, PrintStream ps) throws AlgebricksException {
                 try {
-                    if (typedItemList) {
-                        for (int i = 0; i < numberOfitems; i++) {
-                            itemLength = NonTaggedFormatUtil.getFieldValueLength(b, itemOffset, itemTag, false);
-                            itemPrinter.print(b, itemOffset - 1, itemLength, ps);
-                            itemOffset += itemLength;
-                            if (i + 1 < numberOfitems)
-                                ps.print(", ");
-                        }
-                    } else {
-                        for (int i = 0; i < numberOfitems; i++) {
-                            itemTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b[itemOffset]);
-                            itemLength = NonTaggedFormatUtil.getFieldValueLength(b, itemOffset, itemTag, true) + 1;
-                            itemPrinter.print(b, itemOffset, itemLength, ps);
-                            itemOffset += itemLength;
-                            if (i + 1 < numberOfitems)
-                                ps.print(", ");
-                        }
-                    }
-                } catch (AsterixException e) {
-                    throw new AlgebricksException(e);
+                    listAccessor.set(b, start, l);
+                    arg.first = ps;
+                    listAccessor.accept(printVisitor, arg);
+                } catch (Exception ioe) {
+                    throw new AlgebricksException(ioe);
                 }
-                ps.print(" ]");
             }
         };
     }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/APoint3DPrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/APoint3DPrinter.java
index 699cb77..7394dc5 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/APoint3DPrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/APoint3DPrinter.java
@@ -8,7 +8,6 @@
 
 public class APoint3DPrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final APoint3DPrinter INSTANCE = new APoint3DPrinter();
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/APointPrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/APointPrinter.java
index 50fbb0b..7c50905 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/APointPrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/APointPrinter.java
@@ -8,7 +8,6 @@
 
 public class APointPrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final APointPrinter INSTANCE = new APointPrinter();
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/APolygonPrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/APolygonPrinter.java
index 794dae7..bc1e553 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/APolygonPrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/APolygonPrinter.java
@@ -9,7 +9,6 @@
 
 public class APolygonPrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final APolygonPrinter INSTANCE = new APolygonPrinter();
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ARecordPrinterFactory.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ARecordPrinterFactory.java
index fafe320..e4b2676 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ARecordPrinterFactory.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ARecordPrinterFactory.java
@@ -2,23 +2,21 @@
 
 import java.io.PrintStream;
 
-import edu.uci.ics.asterix.common.exceptions.AsterixException;
-import edu.uci.ics.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
-import edu.uci.ics.asterix.formats.nontagged.AqlPrinterFactoryProvider;
+import edu.uci.ics.asterix.om.pointables.PointableAllocator;
+import edu.uci.ics.asterix.om.pointables.base.DefaultOpenFieldType;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.pointables.printer.APrintVisitor;
 import edu.uci.ics.asterix.om.types.ARecordType;
 import edu.uci.ics.asterix.om.types.ATypeTag;
-import edu.uci.ics.asterix.om.types.AUnionType;
-import edu.uci.ics.asterix.om.types.EnumDeserializer;
 import edu.uci.ics.asterix.om.types.IAType;
-import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
 import edu.uci.ics.hyracks.algebricks.data.IPrinter;
 import edu.uci.ics.hyracks.algebricks.data.IPrinterFactory;
 
 public class ARecordPrinterFactory implements IPrinterFactory {
 
     private static final long serialVersionUID = 1L;
-
     private final ARecordType recType;
 
     public ARecordPrinterFactory(ARecordType recType) {
@@ -28,137 +26,29 @@
     @Override
     public IPrinter createPrinter() {
 
-        return new IPrinter() {
+        PointableAllocator allocator = new PointableAllocator();
+        final IAType inputType = recType == null ? DefaultOpenFieldType.getDefaultOpenFieldType(ATypeTag.RECORD)
+                : recType;
+        final IVisitablePointable recAccessor = allocator.allocateRecordValue(inputType);
+        final APrintVisitor printVisitor = new APrintVisitor();
+        final Pair<PrintStream, ATypeTag> arg = new Pair<PrintStream, ATypeTag>(null, null);
 
-            private IPrinter[] fieldPrinters;
-            private IAType[] fieldTypes;
-            private int[] fieldOffsets;
-            private int numberOfSchemaFields, numberOfOpenFields, openPartOffset, fieldOffset, offsetArrayOffset,
-                    fieldValueLength, nullBitMapOffset, recordOffset;
-            private boolean isExpanded, hasNullableFields;
-            private ATypeTag tag;
+        return new IPrinter() {
 
             @Override
             public void init() throws AlgebricksException {
-
-                numberOfSchemaFields = 0;
-                if (recType != null) {
-                    numberOfSchemaFields = recType.getFieldNames().length;
-                    fieldPrinters = new IPrinter[numberOfSchemaFields];
-                    fieldTypes = new IAType[numberOfSchemaFields];
-                    fieldOffsets = new int[numberOfSchemaFields];
-                    for (int i = 0; i < numberOfSchemaFields; i++) {
-                        fieldTypes[i] = recType.getFieldTypes()[i];
-                        if (fieldTypes[i].getTypeTag() == ATypeTag.UNION
-                                && NonTaggedFormatUtil.isOptionalField((AUnionType) fieldTypes[i]))
-                            fieldPrinters[i] = (AqlPrinterFactoryProvider.INSTANCE
-                                    .getPrinterFactory(((AUnionType) fieldTypes[i]).getUnionList().get(
-                                            NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST))).createPrinter();
-                        else
-                            fieldPrinters[i] = (AqlPrinterFactoryProvider.INSTANCE.getPrinterFactory(fieldTypes[i]))
-                                    .createPrinter();
-                        fieldPrinters[i].init();
-                    }
-                }
+                arg.second = inputType.getTypeTag();
             }
 
             @Override
             public void print(byte[] b, int start, int l, PrintStream ps) throws AlgebricksException {
-                ps.print("{ ");
-                isExpanded = false;
-                openPartOffset = 0;
-                int s = start;
-                recordOffset = s;
-                if (recType == null) {
-                    openPartOffset = s + AInt32SerializerDeserializer.getInt(b, s + 6);
-                    s += 8;
-                    isExpanded = true;
-                } else {
-                    if (recType.isOpen()) {
-                        isExpanded = b[s + 5] == 1 ? true : false;
-                        if (isExpanded) {
-                            openPartOffset = s + AInt32SerializerDeserializer.getInt(b, s + 6);
-                            s += 10;
-                        } else
-                            s += 6;
-                    } else
-                        s += 5;
-                }
                 try {
-                    if (numberOfSchemaFields > 0) {
-                        s += 4;
-                        nullBitMapOffset = 0;
-                        hasNullableFields = NonTaggedFormatUtil.hasNullableField(recType);
-                        if (hasNullableFields) {
-                            nullBitMapOffset = s;
-                            offsetArrayOffset = s
-                                    + (this.numberOfSchemaFields % 8 == 0 ? numberOfSchemaFields / 8
-                                            : numberOfSchemaFields / 8 + 1);
-                        } else {
-                            offsetArrayOffset = s;
-                        }
-                        for (int i = 0; i < numberOfSchemaFields; i++) {
-                            fieldOffsets[i] = AInt32SerializerDeserializer.getInt(b, offsetArrayOffset) + recordOffset;
-                            offsetArrayOffset += 4;
-                        }
-                        for (int fieldNumber = 0; fieldNumber < numberOfSchemaFields; fieldNumber++) {
-                            if (fieldNumber != 0) {
-                                ps.print(", ");
-                            }
-                            ps.print("\"");
-                            ps.print(recType.getFieldNames()[fieldNumber]);
-                            ps.print("\": ");
-                            if (hasNullableFields) {
-                                byte b1 = b[nullBitMapOffset + fieldNumber / 8];
-                                int p = 1 << (7 - (fieldNumber % 8));
-                                if ((b1 & p) == 0) {
-                                    ps.print("null");
-                                    continue;
-                                }
-                            }
-                            if (fieldTypes[fieldNumber].getTypeTag() == ATypeTag.UNION) {
-                                if (NonTaggedFormatUtil.isOptionalField((AUnionType) fieldTypes[fieldNumber])) {
-                                    tag = ((AUnionType) fieldTypes[fieldNumber]).getUnionList()
-                                            .get(NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST).getTypeTag();
-                                    fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b,
-                                            fieldOffsets[fieldNumber], tag, false);
-                                    fieldPrinters[fieldNumber].print(b, fieldOffsets[fieldNumber] - 1,
-                                            fieldValueLength, ps);
-                                }
-                            } else {
-                                tag = fieldTypes[fieldNumber].getTypeTag();
-                                fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b,
-                                        fieldOffsets[fieldNumber], tag, false);
-                                fieldPrinters[fieldNumber]
-                                        .print(b, fieldOffsets[fieldNumber] - 1, fieldValueLength, ps);
-                            }
-                        }
-                        if (isExpanded)
-                            ps.print(", ");
-                    }
-                    if (isExpanded) {
-                        numberOfOpenFields = AInt32SerializerDeserializer.getInt(b, openPartOffset);
-                        fieldOffset = openPartOffset + 4 + (8 * numberOfOpenFields);
-                        for (int i = 0; i < numberOfOpenFields; i++) {
-                            // we print the field name
-                            fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b, fieldOffset, ATypeTag.STRING,
-                                    false);
-                            AStringPrinter.INSTANCE.print(b, fieldOffset - 1, fieldValueLength, ps);
-                            fieldOffset += fieldValueLength;
-                            ps.print(": ");
-                            // now we print the value
-                            tag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b[fieldOffset]);
-                            fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b, fieldOffset, tag, true) + 1;
-                            AObjectPrinter.INSTANCE.print(b, fieldOffset, fieldValueLength, ps);
-                            fieldOffset += fieldValueLength;
-                            if (i + 1 < numberOfOpenFields)
-                                ps.print(", ");
-                        }
-                    }
-                } catch (AsterixException e) {
-                    throw new AlgebricksException(e);
+                    recAccessor.set(b, start, l);
+                    arg.first = ps;
+                    recAccessor.accept(printVisitor, arg);
+                } catch (Exception ioe) {
+                    throw new AlgebricksException(ioe);
                 }
-                ps.print(" }");
             }
         };
     }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ARectanglePrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ARectanglePrinter.java
index 17caa53..0618f28 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ARectanglePrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ARectanglePrinter.java
@@ -8,7 +8,6 @@
 
 public class ARectanglePrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final ARectanglePrinter INSTANCE = new ARectanglePrinter();
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AStringPrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AStringPrinter.java
index 6f5b629..d491777 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AStringPrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AStringPrinter.java
@@ -9,7 +9,6 @@
 
 public class AStringPrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final AStringPrinter INSTANCE = new AStringPrinter();
 
     @Override
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ATimePrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ATimePrinter.java
index 72a6e37..b77ee64 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ATimePrinter.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/ATimePrinter.java
@@ -9,10 +9,9 @@
 
 public class ATimePrinter implements IPrinter {
 
-    private static final long serialVersionUID = 1L;
     public static final ATimePrinter INSTANCE = new ATimePrinter();
     private static final GregorianCalendarSystem gCalInstance = GregorianCalendarSystem.getInstance();
-    
+
     @Override
     public void init() {
 
@@ -21,7 +20,7 @@
     @Override
     public void print(byte[] b, int s, int l, PrintStream ps) throws AlgebricksException {
         int time = AInt32SerializerDeserializer.getInt(b, s + 1);
-        
+
         ps.print("time(\"");
         ps.append(String.format("%02d", gCalInstance.getHourOfDay(time))).append(":")
                 .append(String.format("%02d", gCalInstance.getMinOfHour(time))).append(":")
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AUnorderedlistPrinterFactory.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AUnorderedlistPrinterFactory.java
index 59af365..f779919 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AUnorderedlistPrinterFactory.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/printers/AUnorderedlistPrinterFactory.java
@@ -2,22 +2,21 @@
 
 import java.io.PrintStream;
 
-import edu.uci.ics.asterix.common.exceptions.AsterixException;
-import edu.uci.ics.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
-import edu.uci.ics.asterix.formats.nontagged.AqlPrinterFactoryProvider;
+import edu.uci.ics.asterix.om.pointables.PointableAllocator;
+import edu.uci.ics.asterix.om.pointables.base.DefaultOpenFieldType;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.pointables.printer.APrintVisitor;
 import edu.uci.ics.asterix.om.types.ATypeTag;
 import edu.uci.ics.asterix.om.types.AUnorderedListType;
-import edu.uci.ics.asterix.om.types.EnumDeserializer;
 import edu.uci.ics.asterix.om.types.IAType;
-import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
 import edu.uci.ics.hyracks.algebricks.data.IPrinter;
 import edu.uci.ics.hyracks.algebricks.data.IPrinterFactory;
 
 public class AUnorderedlistPrinterFactory implements IPrinterFactory {
 
     private static final long serialVersionUID = 1L;
-
     private AUnorderedListType unorderedlistType;
 
     public AUnorderedlistPrinterFactory(AUnorderedListType unorderedlistType) {
@@ -27,77 +26,29 @@
     @Override
     public IPrinter createPrinter() {
 
-        return new IPrinter() {
+        PointableAllocator allocator = new PointableAllocator();
+        final IAType inputType = unorderedlistType == null ? DefaultOpenFieldType
+                .getDefaultOpenFieldType(ATypeTag.UNORDEREDLIST) : unorderedlistType;
+        final IVisitablePointable listAccessor = allocator.allocateListValue(inputType);
+        final APrintVisitor printVisitor = new APrintVisitor();
+        final Pair<PrintStream, ATypeTag> arg = new Pair<PrintStream, ATypeTag>(null, null);
 
-            private IPrinter itemPrinter;
-            private IAType itemType;
-            private ATypeTag itemTag;
-            private boolean typedItemList = false;
+        return new IPrinter() {
 
             @Override
             public void init() throws AlgebricksException {
-
-                if (unorderedlistType != null && unorderedlistType.getItemType() != null) {
-                    itemType = unorderedlistType.getItemType();
-                    if (itemType.getTypeTag() == ATypeTag.ANY) {
-                        this.typedItemList = false;
-                        this.itemPrinter = AObjectPrinterFactory.INSTANCE.createPrinter();
-                    } else {
-                        this.typedItemList = true;
-                        itemPrinter = AqlPrinterFactoryProvider.INSTANCE.getPrinterFactory(itemType).createPrinter();
-                        itemTag = unorderedlistType.getItemType().getTypeTag();
-                    }
-                } else {
-                    this.typedItemList = false;
-                    this.itemPrinter = AObjectPrinterFactory.INSTANCE.createPrinter();
-                }
-                itemPrinter.init();
+                arg.second = inputType.getTypeTag();
             }
 
             @Override
-            public void print(byte[] b, int s, int l, PrintStream ps) throws AlgebricksException {
-                ps.print("{{ ");
-                int numberOfitems = AInt32SerializerDeserializer.getInt(b, s + 6);
-                int itemOffset;
-                if (typedItemList) {
-                    switch (itemTag) {
-                        case STRING:
-                        case RECORD:
-                        case ORDEREDLIST:
-                        case UNORDEREDLIST:
-                        case ANY:
-                            itemOffset = s + 10 + (numberOfitems * 4);
-                            break;
-                        default:
-                            itemOffset = s + 10;
-                    }
-                } else
-                    itemOffset = s + 10 + (numberOfitems * 4);
-                int itemLength;
-
+            public void print(byte[] b, int start, int l, PrintStream ps) throws AlgebricksException {
                 try {
-                    if (typedItemList) {
-                        for (int i = 0; i < numberOfitems; i++) {
-                            itemLength = NonTaggedFormatUtil.getFieldValueLength(b, itemOffset, itemTag, false);
-                            itemPrinter.print(b, itemOffset - 1, itemLength, ps);
-                            itemOffset += itemLength;
-                            if (i + 1 < numberOfitems)
-                                ps.print(", ");
-                        }
-                    } else {
-                        for (int i = 0; i < numberOfitems; i++) {
-                            itemTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b[itemOffset]);
-                            itemLength = NonTaggedFormatUtil.getFieldValueLength(b, itemOffset, itemTag, true) + 1;
-                            itemPrinter.print(b, itemOffset, itemLength, ps);
-                            itemOffset += itemLength;
-                            if (i + 1 < numberOfitems)
-                                ps.print(", ");
-                        }
-                    }
-                } catch (AsterixException e) {
-                    throw new AlgebricksException(e);
+                    listAccessor.set(b, start, l);
+                    arg.first = ps;
+                    listAccessor.accept(printVisitor, arg);
+                } catch (Exception ioe) {
+                    throw new AlgebricksException(ioe);
                 }
-                ps.print(" }}");
             }
         };
     }
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/serde/AOrderedListSerializerDeserializer.java b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/serde/AOrderedListSerializerDeserializer.java
index c68f82a..d37a4ea 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/serde/AOrderedListSerializerDeserializer.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/dataflow/data/nontagged/serde/AOrderedListSerializerDeserializer.java
@@ -26,9 +26,9 @@
     public static final AOrderedListSerializerDeserializer SCHEMALESS_INSTANCE = new AOrderedListSerializerDeserializer();
 
     private IAType itemType;
-    @SuppressWarnings("unchecked")
+    @SuppressWarnings("rawtypes")
     private ISerializerDeserializer serializer;
-    @SuppressWarnings("unchecked")
+    @SuppressWarnings("rawtypes")
     private ISerializerDeserializer deserializer;
     private AOrderedListType orderedlistType;
 
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/formats/nontagged/AqlSerializerDeserializerProvider.java b/asterix-om/src/main/java/edu/uci/ics/asterix/formats/nontagged/AqlSerializerDeserializerProvider.java
index d4d4fb6..5492c5c 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/formats/nontagged/AqlSerializerDeserializerProvider.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/formats/nontagged/AqlSerializerDeserializerProvider.java
@@ -48,7 +48,7 @@
     private AqlSerializerDeserializerProvider() {
     }
 
-    @SuppressWarnings("unchecked")
+    @SuppressWarnings("rawtypes")
     @Override
     public ISerializerDeserializer getSerializerDeserializer(Object typeInfo) {
         IAType aqlType = (IAType) typeInfo;
@@ -63,7 +63,7 @@
         }
     }
 
-    @SuppressWarnings("unchecked")
+    @SuppressWarnings("rawtypes")
     public ISerializerDeserializer getNonTaggedSerializerDeserializer(IAType aqlType) {
         switch (aqlType.getTypeTag()) {
             case CIRCLE: {
@@ -139,7 +139,7 @@
         }
     }
 
-    @SuppressWarnings("unchecked")
+    @SuppressWarnings("rawtypes")
     private ISerializerDeserializer addTag(final ISerializerDeserializer nonTaggedSerde, final ATypeTag typeTag) {
         return new ISerializerDeserializer<IAObject>() {
 
@@ -148,13 +148,15 @@
             @Override
             public IAObject deserialize(DataInput in) throws HyracksDataException {
                 try {
-                    ATypeTag typeTag = SerializerDeserializerUtil.deserializeTag(in);
+                    //deserialize the tag to move the  input cursor forward 
+                    SerializerDeserializerUtil.deserializeTag(in);
                 } catch (IOException e) {
                     throw new HyracksDataException(e);
                 }
                 return (IAObject) nonTaggedSerde.deserialize(in);
             }
 
+            @SuppressWarnings("unchecked")
             @Override
             public void serialize(IAObject instance, DataOutput out) throws HyracksDataException {
                 SerializerDeserializerUtil.serializeTag(instance, out);
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/AFlatValuePointable.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/AFlatValuePointable.java
new file mode 100644
index 0000000..445e2a2
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/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.om.pointables;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.pointables.visitor.IVisitablePointableVisitor;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.container.IObjectFactory;
+import edu.uci.ics.hyracks.data.std.api.IValueReference;
+
+/**
+ * This class represents a flat field, e.g., int field, string field, and
+ * so on, based on a 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 IObjectFactory<IVisitablePointable, IAType> FACTORY = new IObjectFactory<IVisitablePointable, IAType>() {
+        public AFlatValuePointable create(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-om/src/main/java/edu/uci/ics/asterix/om/pointables/AListPointable.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/AListPointable.java
new file mode 100644
index 0000000..ab32b6b
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/AListPointable.java
@@ -0,0 +1,179 @@
+/*
+ * 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.om.pointables;
+
+import java.io.DataOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.pointables.visitor.IVisitablePointableVisitor;
+import edu.uci.ics.asterix.om.types.AOrderedListType;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+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.om.util.NonTaggedFormatUtil;
+import edu.uci.ics.asterix.om.util.ResettableByteArrayOutputStream;
+import edu.uci.ics.asterix.om.util.container.IObjectFactory;
+
+/**
+ * This class interprets 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 {
+
+    /**
+     * DO NOT allow to create AListPointable object arbitrarily, force to use
+     * object pool based allocator, in order to have object reuse.
+     */
+    static IObjectFactory<IVisitablePointable, IAType> FACTORY = new IObjectFactory<IVisitablePointable, IAType>() {
+        public IVisitablePointable create(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 boolean ordered = false;
+
+    /**
+     * private constructor, to prevent constructing it arbitrarily
+     * 
+     * @param inputType
+     */
+    private AListPointable(AbstractCollectionType inputType) {
+        if (inputType instanceof AOrderedListType) {
+            ordered = true;
+        }
+        if (inputType != null && inputType.getItemType() != null) {
+            itemType = inputType.getItemType();
+            if (itemType.getTypeTag() == ATypeTag.ANY) {
+                typedItemList = false;
+            } else {
+                typedItemList = true;
+                itemTag = inputType.getItemType().getTypeTag();
+            }
+        } else {
+            this.typedItemList = false;
+        }
+    }
+
+    private void reset() {
+        allocator.reset();
+        items.clear();
+        itemTags.clear();
+        dataBos.reset();
+    }
+
+    @Override
+    public void set(byte[] b, int s, int len) {
+        reset();
+
+        int numberOfitems = AInt32SerializerDeserializer.getInt(b, s + 6);
+        int itemOffset;
+        if (typedItemList) {
+            switch (itemTag) {
+                case STRING:
+                case RECORD:
+                case ORDEREDLIST:
+                case UNORDEREDLIST:
+                case ANY:
+                    itemOffset = s + 10 + (numberOfitems * 4);
+                    break;
+                default:
+                    itemOffset = s + 10;
+            }
+        } 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);
+                    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.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.set(dataBos.getByteArray(), start, end - start);
+                    itemOffset += itemLength;
+                    items.add(item);
+                }
+            } else {
+                for (int i = 0; i < numberOfitems; i++) {
+                    itemTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b[itemOffset]);
+                    itemLength = NonTaggedFormatUtil.getFieldValueLength(b, itemOffset, itemTag, true) + 1;
+                    IVisitablePointable tag = allocator.allocateEmpty();
+                    IVisitablePointable item = allocator.allocateFieldValue(itemTag);
+
+                    // set item type tag
+                    int start = dataBos.size();
+                    dataDos.writeByte(itemTag.serialize());
+                    int end = dataBos.size();
+                    tag.set(dataBos.getByteArray(), start, end - start);
+                    itemTags.add(tag);
+
+                    // open part field already include the type tag
+                    item.set(b, itemOffset, itemLength);
+                    itemOffset += itemLength;
+                    items.add(item);
+                }
+            }
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+    }
+
+    @Override
+    public <R, T> R accept(IVisitablePointableVisitor<R, T> vistor, T tag) throws AsterixException {
+        return vistor.visit(this, tag);
+    }
+
+    public List<IVisitablePointable> getItems() {
+        return items;
+    }
+
+    public List<IVisitablePointable> getItemTags() {
+        return itemTags;
+    }
+
+    public boolean ordered() {
+        return ordered;
+    }
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/ARecordPointable.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/ARecordPointable.java
new file mode 100644
index 0000000..7e097af
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/ARecordPointable.java
@@ -0,0 +1,285 @@
+/*
+ * 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.om.pointables;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.dataflow.data.nontagged.AqlNullWriterFactory;
+import edu.uci.ics.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.pointables.visitor.IVisitablePointableVisitor;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.AUnionType;
+import edu.uci.ics.asterix.om.types.EnumDeserializer;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
+import edu.uci.ics.asterix.om.util.ResettableByteArrayOutputStream;
+import edu.uci.ics.asterix.om.util.container.IObjectFactory;
+import edu.uci.ics.hyracks.api.dataflow.value.INullWriter;
+
+/**
+ * This class interprets 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 {
+
+    /**
+     * DO NOT allow to create ARecordPointable object arbitrarily, force to use
+     * object pool based allocator, in order to have object reuse
+     */
+    static IObjectFactory<IVisitablePointable, IAType> FACTORY = new IObjectFactory<IVisitablePointable, IAType>() {
+        public IVisitablePointable create(IAType type) {
+            return new ARecordPointable((ARecordType) type);
+        }
+    };
+
+    // access results: field names, field types, and field values
+    private final List<IVisitablePointable> fieldNames = new ArrayList<IVisitablePointable>();
+    private final List<IVisitablePointable> fieldTypeTags = new ArrayList<IVisitablePointable>();
+    private final List<IVisitablePointable> fieldValues = new ArrayList<IVisitablePointable>();
+
+    // pointable allocator
+    private final PointableAllocator allocator = new PointableAllocator();
+
+    private final ResettableByteArrayOutputStream typeBos = new ResettableByteArrayOutputStream();
+    private final DataOutputStream typeDos = new DataOutputStream(typeBos);
+
+    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.create(null);
+
+    private int closedPartTypeInfoSize = 0;
+    private int offsetArrayOffset;
+    private ATypeTag typeTag;
+
+    /**
+     * private constructor, to prevent constructing it arbitrarily
+     * 
+     * @param inputType
+     */
+    private ARecordPointable(ARecordType inputType) {
+        this.inputRecType = inputType;
+        IAType[] fieldTypes = inputType.getFieldTypes();
+        String[] fieldNameStrs = inputType.getFieldNames();
+        numberOfSchemaFields = fieldTypes.length;
+
+        // initialize the buffer for closed parts(fieldName bytes+ type bytes) +
+        // constant(null bytes)
+        typeBos.reset();
+        try {
+            for (int i = 0; i < numberOfSchemaFields; i++) {
+                ATypeTag ftypeTag = fieldTypes[i].getTypeTag();
+
+                if (fieldTypes[i].getTypeTag() == ATypeTag.UNION
+                        && NonTaggedFormatUtil.isOptionalField((AUnionType) fieldTypes[i]))
+                    // optional field: add the embedded non-null type tag
+                    ftypeTag = ((AUnionType) fieldTypes[i]).getUnionList()
+                            .get(NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST).getTypeTag();
+
+                // add type tag Reference
+                int tagStart = typeBos.size();
+                typeDos.writeByte(ftypeTag.serialize());
+                int tagEnd = typeBos.size();
+                IVisitablePointable typeTagReference = AFlatValuePointable.FACTORY.create(null);
+                typeTagReference.set(typeBos.getByteArray(), tagStart, tagEnd - tagStart);
+                fieldTypeTags.add(typeTagReference);
+
+                // add type name Reference (including a astring type tag)
+                int nameStart = typeBos.size();
+                typeDos.writeByte(ATypeTag.STRING.serialize());
+                typeDos.writeUTF(fieldNameStrs[i]);
+                int nameEnd = typeBos.size();
+                IVisitablePointable typeNameReference = AFlatValuePointable.FACTORY.create(null);
+                typeNameReference.set(typeBos.getByteArray(), nameStart, nameEnd - nameStart);
+                fieldNames.add(typeNameReference);
+            }
+
+            // initialize a constant: null value bytes reference
+            int nullFieldStart = typeBos.size();
+            INullWriter nullWriter = AqlNullWriterFactory.INSTANCE.createNullWriter();
+            nullWriter.writeNull(typeDos);
+            int nullFieldEnd = typeBos.size();
+            nullReference.set(typeBos.getByteArray(), nullFieldStart, nullFieldEnd - nullFieldStart);
+        } catch (IOException e) {
+            throw new IllegalStateException(e);
+        }
+        closedPartTypeInfoSize = typeBos.size();
+        fieldOffsets = new int[numberOfSchemaFields];
+    }
+
+    private void reset() {
+        typeBos.reset(closedPartTypeInfoSize);
+        dataBos.reset(0);
+        // reset the allocator
+        allocator.reset();
+
+        // clean up the returned containers
+        for (int i = fieldNames.size() - 1; i >= numberOfSchemaFields; i--)
+            fieldNames.remove(i);
+        for (int i = fieldTypeTags.size() - 1; i >= numberOfSchemaFields; i--)
+            fieldTypeTags.remove(i);
+        fieldValues.clear();
+    }
+
+    @Override
+    public void set(byte[] b, int start, int len) {
+        // clear the previous states
+        reset();
+        super.set(b, start, len);
+
+        boolean isExpanded = false;
+        int openPartOffset = 0;
+        int s = start;
+        int recordOffset = s;
+        if (inputRecType == null) {
+            openPartOffset = s + AInt32SerializerDeserializer.getInt(b, s + 6);
+            s += 8;
+            isExpanded = true;
+        } else {
+            if (inputRecType.isOpen()) {
+                isExpanded = b[s + 5] == 1 ? true : false;
+                if (isExpanded) {
+                    openPartOffset = s + AInt32SerializerDeserializer.getInt(b, s + 6);
+                    s += 10;
+                } else {
+                    s += 6;
+                }
+            } else {
+                s += 5;
+            }
+        }
+        try {
+            if (numberOfSchemaFields > 0) {
+                s += 4;
+                int nullBitMapOffset = 0;
+                boolean hasNullableFields = NonTaggedFormatUtil.hasNullableField(inputRecType);
+                if (hasNullableFields) {
+                    nullBitMapOffset = s;
+                    offsetArrayOffset = s
+                            + (this.numberOfSchemaFields % 8 == 0 ? numberOfSchemaFields / 8
+                                    : numberOfSchemaFields / 8 + 1);
+                } else {
+                    offsetArrayOffset = s;
+                }
+                for (int i = 0; i < numberOfSchemaFields; i++) {
+                    fieldOffsets[i] = AInt32SerializerDeserializer.getInt(b, offsetArrayOffset) + recordOffset;
+                    offsetArrayOffset += 4;
+                }
+                for (int fieldNumber = 0; fieldNumber < numberOfSchemaFields; fieldNumber++) {
+                    if (hasNullableFields) {
+                        byte b1 = b[nullBitMapOffset + fieldNumber / 8];
+                        int p = 1 << (7 - (fieldNumber % 8));
+                        if ((b1 & p) == 0) {
+                            // set null value (including type tag inside)
+                            fieldValues.add(nullReference);
+                            continue;
+                        }
+                    }
+                    IAType[] fieldTypes = inputRecType.getFieldTypes();
+                    int fieldValueLength = 0;
+
+                    IAType fieldType = fieldTypes[fieldNumber];
+                    if (fieldTypes[fieldNumber].getTypeTag() == ATypeTag.UNION) {
+                        if (NonTaggedFormatUtil.isOptionalField((AUnionType) fieldTypes[fieldNumber])) {
+                            fieldType = ((AUnionType) fieldTypes[fieldNumber]).getUnionList().get(
+                                    NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST);
+                            typeTag = fieldType.getTypeTag();
+                            fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b, fieldOffsets[fieldNumber],
+                                    typeTag, false);
+                        }
+                    } else {
+                        typeTag = fieldTypes[fieldNumber].getTypeTag();
+                        fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b, fieldOffsets[fieldNumber],
+                                typeTag, false);
+                    }
+                    // set field value (including the type tag)
+                    int fstart = dataBos.size();
+                    dataDos.writeByte(typeTag.serialize());
+                    dataDos.write(b, fieldOffsets[fieldNumber], fieldValueLength);
+                    int fend = dataBos.size();
+                    IVisitablePointable fieldValue = allocator.allocateFieldValue(fieldType);
+                    fieldValue.set(dataBos.getByteArray(), fstart, fend - fstart);
+                    fieldValues.add(fieldValue);
+                }
+            }
+            if (isExpanded) {
+                int numberOfOpenFields = AInt32SerializerDeserializer.getInt(b, openPartOffset);
+                int fieldOffset = openPartOffset + 4 + (8 * numberOfOpenFields);
+                for (int i = 0; i < numberOfOpenFields; i++) {
+                    // set the field name (including a type tag, which is
+                    // astring)
+                    int fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b, fieldOffset, ATypeTag.STRING,
+                            false);
+                    int fnstart = dataBos.size();
+                    dataDos.writeByte(ATypeTag.STRING.serialize());
+                    dataDos.write(b, fieldOffset, fieldValueLength);
+                    int fnend = dataBos.size();
+                    IVisitablePointable fieldName = allocator.allocateEmpty();
+                    fieldName.set(dataBos.getByteArray(), fnstart, fnend - fnstart);
+                    fieldNames.add(fieldName);
+                    fieldOffset += fieldValueLength;
+
+                    // set the field type tag
+                    IVisitablePointable fieldTypeTag = allocator.allocateEmpty();
+                    fieldTypeTag.set(b, fieldOffset, 1);
+                    fieldTypeTags.add(fieldTypeTag);
+                    typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b[fieldOffset]);
+
+                    // set the field value (already including type tag)
+                    fieldValueLength = NonTaggedFormatUtil.getFieldValueLength(b, fieldOffset, typeTag, true) + 1;
+
+                    // allocate
+                    IVisitablePointable fieldValueAccessor = allocator.allocateFieldValue(typeTag);
+                    fieldValueAccessor.set(b, fieldOffset, fieldValueLength);
+                    fieldValues.add(fieldValueAccessor);
+                    fieldOffset += fieldValueLength;
+                }
+            }
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+    }
+
+    public List<IVisitablePointable> getFieldNames() {
+        return fieldNames;
+    }
+
+    public List<IVisitablePointable> getFieldTypeTags() {
+        return fieldTypeTags;
+    }
+
+    public List<IVisitablePointable> getFieldValues() {
+        return fieldValues;
+    }
+
+    @Override
+    public <R, T> R accept(IVisitablePointableVisitor<R, T> vistor, T tag) throws AsterixException {
+        return vistor.visit(this, tag);
+    }
+
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/AbstractVisitablePointable.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/AbstractVisitablePointable.java
new file mode 100644
index 0000000..943a967
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/AbstractVisitablePointable.java
@@ -0,0 +1,58 @@
+/*
+ * 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.om.pointables;
+
+import edu.uci.ics.asterix.om.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 = -1;
+    private int len = -1;
+
+    @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-om/src/main/java/edu/uci/ics/asterix/om/pointables/PointableAllocator.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/PointableAllocator.java
new file mode 100644
index 0000000..5ad2ccb
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/PointableAllocator.java
@@ -0,0 +1,91 @@
+/*
+ * 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.om.pointables;
+
+import edu.uci.ics.asterix.om.pointables.base.DefaultOpenFieldType;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.container.IObjectPool;
+import edu.uci.ics.asterix.om.util.container.ListObjectPool;
+
+/**
+ * This class is the ONLY place to create IVisitablePointable object instances,
+ * to enforce use of an object pool.
+ */
+public class PointableAllocator {
+
+    private IObjectPool<IVisitablePointable, IAType> flatValueAllocator = new ListObjectPool<IVisitablePointable, IAType>(
+            AFlatValuePointable.FACTORY);
+    private IObjectPool<IVisitablePointable, IAType> recordValueAllocator = new ListObjectPool<IVisitablePointable, IAType>(
+            ARecordPointable.FACTORY);
+    private IObjectPool<IVisitablePointable, IAType> listValueAllocator = new ListObjectPool<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-om/src/main/java/edu/uci/ics/asterix/om/pointables/base/DefaultOpenFieldType.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/base/DefaultOpenFieldType.java
new file mode 100644
index 0000000..9184616
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/base/DefaultOpenFieldType.java
@@ -0,0 +1,57 @@
+/*
+ * 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.om.pointables.base;
+
+import edu.uci.ics.asterix.om.types.AOrderedListType;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.AUnorderedListType;
+import edu.uci.ics.asterix.om.types.BuiltinType;
+import edu.uci.ics.asterix.om.types.IAType;
+
+/**
+ * This class serves as the repository for the 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
+    public static ARecordType NESTED_OPEN_RECORD_TYPE = new ARecordType("nested-open", new String[] {},
+            new IAType[] {}, true);
+
+    // nested open list type
+    public static AOrderedListType NESTED_OPEN_AORDERED_LIST_TYPE = new AOrderedListType(BuiltinType.ANY,
+            "nested-ordered-list");
+
+    // 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))
+            return NESTED_OPEN_RECORD_TYPE;
+        if (tag.equals(ATypeTag.ORDEREDLIST))
+            return NESTED_OPEN_AORDERED_LIST_TYPE;
+        if (tag.equals(ATypeTag.UNORDEREDLIST))
+            return NESTED_OPEN_AUNORDERED_LIST_TYPE;
+        else
+            return null;
+    }
+
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/base/IVisitablePointable.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/base/IVisitablePointable.java
new file mode 100644
index 0000000..7456ee8
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/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.om.pointables.base;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.om.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-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ACastVisitor.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ACastVisitor.java
new file mode 100644
index 0000000..9fccf4a
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ACastVisitor.java
@@ -0,0 +1,95 @@
+/*
+ * 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.om.pointables.cast;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.om.pointables.AFlatValuePointable;
+import edu.uci.ics.asterix.om.pointables.AListPointable;
+import edu.uci.ics.asterix.om.pointables.ARecordPointable;
+import edu.uci.ics.asterix.om.pointables.base.DefaultOpenFieldType;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.pointables.visitor.IVisitablePointableVisitor;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.AbstractCollectionType;
+import edu.uci.ics.asterix.om.types.IAType;
+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 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/closed part of a record has a 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 {
+            if (arg.second.getTypeTag().equals(ATypeTag.ANY)) {
+                arg.second = DefaultOpenFieldType.NESTED_OPEN_AUNORDERED_LIST_TYPE;
+            }
+            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 {
+            if (arg.second.getTypeTag().equals(ATypeTag.ANY)) {
+                arg.second = DefaultOpenFieldType.NESTED_OPEN_RECORD_TYPE;
+            }
+            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-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/AListCaster.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/AListCaster.java
new file mode 100644
index 0000000..c65b3eb
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/AListCaster.java
@@ -0,0 +1,109 @@
+/*
+ * 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.om.pointables.cast;
+
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.List;
+
+import edu.uci.ics.asterix.builders.OrderedListBuilder;
+import edu.uci.ics.asterix.builders.UnorderedListBuilder;
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.om.pointables.AListPointable;
+import edu.uci.ics.asterix.om.pointables.PointableAllocator;
+import edu.uci.ics.asterix.om.pointables.base.DefaultOpenFieldType;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.types.AOrderedListType;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.AUnorderedListType;
+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.om.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.
+ */
+class AListCaster {
+    // pointable allocator
+    private final PointableAllocator allocator = new PointableAllocator();
+
+    // 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 final UnorderedListBuilder unOrderedListBuilder = new UnorderedListBuilder();
+    private final OrderedListBuilder orderedListBuilder = new OrderedListBuilder();
+
+    private final ResettableByteArrayOutputStream dataBos = new ResettableByteArrayOutputStream();
+    private final DataOutput dataDos = new DataOutputStream(dataBos);
+    private IAType reqItemType;
+
+    public AListCaster() {
+
+    }
+
+    public void castList(AListPointable listAccessor, IVisitablePointable resultAccessor,
+            AbstractCollectionType reqType, ACastVisitor visitor) throws IOException, AsterixException {
+        if (reqType.getTypeTag().equals(ATypeTag.UNORDEREDLIST)) {
+            unOrderedListBuilder.reset((AUnorderedListType) reqType);
+            reqItemType = reqType.getItemType();
+        }
+        if (reqType.getTypeTag().equals(ATypeTag.ORDEREDLIST)) {
+            orderedListBuilder.reset((AOrderedListType) reqType);
+            reqItemType = reqType.getItemType();
+        }
+        dataBos.reset();
+
+        List<IVisitablePointable> itemTags = listAccessor.getItemTags();
+        List<IVisitablePointable> items = listAccessor.getItems();
+
+        int start = dataBos.size();
+        for (int i = 0; i < items.size(); i++) {
+            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);
+            } else {
+                if (typeTag != reqItemType.getTypeTag())
+                    throw new AsterixException("mismatched item type");
+                itemVisitorArg.second = reqItemType;
+                item.accept(visitor, itemVisitorArg);
+            }
+            if (reqType.getTypeTag().equals(ATypeTag.ORDEREDLIST)) {
+                orderedListBuilder.addItem(itemVisitorArg.first);
+            }
+            if (reqType.getTypeTag().equals(ATypeTag.UNORDEREDLIST)) {
+                unOrderedListBuilder.addItem(itemVisitorArg.first);
+            }
+        }
+        if (reqType.getTypeTag().equals(ATypeTag.ORDEREDLIST)) {
+            orderedListBuilder.write(dataDos, true);
+        }
+        if (reqType.getTypeTag().equals(ATypeTag.UNORDEREDLIST)) {
+            unOrderedListBuilder.write(dataDos, true);
+        }
+        int end = dataBos.size();
+        resultAccessor.set(dataBos.getByteArray(), start, end - start);
+    }
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ARecordCaster.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ARecordCaster.java
new file mode 100644
index 0000000..fbad7a7
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/cast/ARecordCaster.java
@@ -0,0 +1,318 @@
+/*
+ * 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.om.pointables.cast;
+
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import edu.uci.ics.asterix.builders.RecordBuilder;
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.dataflow.data.nontagged.AqlNullWriterFactory;
+import edu.uci.ics.asterix.om.pointables.ARecordPointable;
+import edu.uci.ics.asterix.om.pointables.PointableAllocator;
+import edu.uci.ics.asterix.om.pointables.base.DefaultOpenFieldType;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.AUnionType;
+import edu.uci.ics.asterix.om.types.EnumDeserializer;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.NonTaggedFormatUtil;
+import edu.uci.ics.asterix.om.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.data.std.util.ByteArrayAccessibleOutputStream;
+
+/**
+ * This class is to do the runtime type cast for a record. It is ONLY visible to
+ * ACastVisitor.
+ */
+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;
+
+    // describe fields (open or not) in the input records
+    private boolean[] openFields;
+    private int[] fieldNamesSortedIndex;
+    private int[] reqFieldNamesSortedIndex;
+
+    public ARecordCaster() {
+        try {
+            bos.reset();
+            int start = bos.size();
+            INullWriter nullWriter = AqlNullWriterFactory.INSTANCE.createNullWriter();
+            nullWriter.writeNull(dos);
+            int end = bos.size();
+            nullReference.set(bos.getByteArray(), start, end - start);
+            start = bos.size();
+            dos.write(ATypeTag.NULL.serialize());
+            end = bos.size();
+            nullTypeTag.set(bos.getByteArray(), start, end);
+        } catch (IOException e) {
+            throw new IllegalStateException(e);
+        }
+    }
+
+    public void castRecord(ARecordPointable recordAccessor, IVisitablePointable resultAccessor, ARecordType reqType,
+            ACastVisitor visitor) throws IOException, AsterixException {
+        List<IVisitablePointable> fieldNames = recordAccessor.getFieldNames();
+        List<IVisitablePointable> fieldTypeTags = recordAccessor.getFieldTypeTags();
+        List<IVisitablePointable> fieldValues = recordAccessor.getFieldValues();
+        numInputFields = fieldNames.size();
+
+        if (openFields == null || numInputFields > openFields.length) {
+            openFields = new boolean[numInputFields];
+            fieldNamesSortedIndex = new int[numInputFields];
+        }
+        if (cachedReqType == null || !reqType.equals(cachedReqType)) {
+            loadRequiredType(reqType);
+        }
+
+        // clear the previous states
+        reset();
+        matchClosedPart(fieldNames, fieldTypeTags, fieldValues);
+        writeOutput(fieldNames, fieldTypeTags, fieldValues, outputDos, visitor);
+        resultAccessor.set(outputBos.getByteArray(), 0, outputBos.size());
+    }
+
+    private void reset() {
+        for (int i = 0; i < numInputFields; i++)
+            openFields[i] = true;
+        for (int i = 0; i < fieldPermutation.length; i++)
+            fieldPermutation[i] = -1;
+        for (int i = 0; i < numInputFields; i++)
+            fieldNamesSortedIndex[i] = i;
+        outputBos.reset();
+    }
+
+    private void loadRequiredType(ARecordType reqType) throws IOException {
+        reqFieldNames.clear();
+        reqFieldTypeTags.clear();
+
+        cachedReqType = reqType;
+        int numSchemaFields = reqType.getFieldTypes().length;
+        IAType[] fieldTypes = reqType.getFieldTypes();
+        String[] fieldNames = reqType.getFieldNames();
+        fieldPermutation = new int[numSchemaFields];
+        optionalFields = new boolean[numSchemaFields];
+        for (int i = 0; i < optionalFields.length; i++)
+            optionalFields[i] = false;
+
+        bos.reset(nullReference.getStartOffset() + nullReference.getLength());
+        for (int i = 0; i < numSchemaFields; i++) {
+            ATypeTag ftypeTag = fieldTypes[i].getTypeTag();
+            String fname = fieldNames[i];
+
+            // add type tag pointable
+            if (fieldTypes[i].getTypeTag() == ATypeTag.UNION
+                    && NonTaggedFormatUtil.isOptionalField((AUnionType) fieldTypes[i])) {
+                // optional field: add the embedded non-null type tag
+                ftypeTag = ((AUnionType) fieldTypes[i]).getUnionList()
+                        .get(NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST).getTypeTag();
+                optionalFields[i] = true;
+            }
+            int tagStart = bos.size();
+            dos.writeByte(ftypeTag.serialize());
+            int tagEnd = bos.size();
+            IVisitablePointable typeTagPointable = allocator.allocateEmpty();
+            typeTagPointable.set(bos.getByteArray(), tagStart, tagEnd - tagStart);
+            reqFieldTypeTags.add(typeTagPointable);
+
+            // add type name pointable (including a string type tag)
+            int nameStart = bos.size();
+            dos.write(ATypeTag.STRING.serialize());
+            dos.writeUTF(fname);
+            int nameEnd = bos.size();
+            IVisitablePointable typeNamePointable = allocator.allocateEmpty();
+            typeNamePointable.set(bos.getByteArray(), nameStart, nameEnd - nameStart);
+            reqFieldNames.add(typeNamePointable);
+        }
+
+        reqFieldNamesSortedIndex = new int[reqFieldNames.size()];
+        for (int i = 0; i < reqFieldNamesSortedIndex.length; i++)
+            reqFieldNamesSortedIndex[i] = i;
+        // sort the field name index
+        quickSort(reqFieldNamesSortedIndex, reqFieldNames, 0, reqFieldNamesSortedIndex.length - 1);
+    }
+
+    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;
+        int reqFnStart = 0;
+        while (fnStart < numInputFields && reqFnStart < reqFieldNames.size()) {
+            int fnPos = fieldNamesSortedIndex[fnStart];
+            int reqFnPos = reqFieldNamesSortedIndex[reqFnStart];
+            int c = compare(fieldNames.get(fnPos), reqFieldNames.get(reqFnPos));
+            if (c == 0) {
+                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))) {
+                    fieldPermutation[reqFnPos] = fnPos;
+                    openFields[fnPos] = false;
+                }
+                fnStart++;
+                reqFnStart++;
+            }
+            if (c > 0)
+                reqFnStart++;
+            if (c < 0)
+                fnStart++;
+        }
+
+        // check unmatched fields in the input type
+        for (int i = 0; i < openFields.length; i++) {
+            if (openFields[i] == true && !cachedReqType.isOpen())
+                throw new IllegalStateException("type mismatch: including extra closed fields");
+        }
+
+        // check unmatched fields in the required type
+        for (int i = 0; i < fieldPermutation.length; i++) {
+            if (fieldPermutation[i] < 0) {
+                IAType t = cachedReqType.getFieldTypes()[i];
+                if (!(t.getTypeTag() == ATypeTag.UNION && NonTaggedFormatUtil.isOptionalField((AUnionType) t))) {
+                    // no matched field in the input for a required closed field
+                    throw new IllegalStateException("type mismatch: miss a required closed field");
+                }
+            }
+        }
+    }
+
+    private void writeOutput(List<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);
+        recBuilder.init();
+
+        // write the closed part
+        for (int i = 0; i < fieldPermutation.length; i++) {
+            int pos = fieldPermutation[i];
+            IVisitablePointable field;
+            if (pos >= 0) {
+                field = fieldValues.get(pos);
+            } else {
+                field = nullReference;
+            }
+            IAType fType = cachedReqType.getFieldTypes()[i];
+            nestedVisitorArg.second = fType;
+
+            // recursively casting, the result of casting can always be thought
+            // as flat
+            if (optionalFields[i]) {
+                nestedVisitorArg.second = ((AUnionType) fType).getUnionList().get(
+                        NonTaggedFormatUtil.OPTIONAL_TYPE_INDEX_IN_UNION_LIST);
+            }
+            field.accept(visitor, nestedVisitorArg);
+            recBuilder.addField(i, nestedVisitorArg.first);
+        }
+
+        // write the open part
+        for (int i = 0; i < numInputFields; i++) {
+            if (openFields[i]) {
+                IVisitablePointable name = fieldNames.get(i);
+                IVisitablePointable field = fieldValues.get(i);
+                IVisitablePointable fieldTypeTag = fieldTypeTags.get(i);
+
+                ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
+                        .deserialize(fieldTypeTag.getByteArray()[fieldTypeTag.getStartOffset()]);
+                nestedVisitorArg.second = DefaultOpenFieldType.getDefaultOpenFieldType(typeTag);
+                field.accept(visitor, nestedVisitorArg);
+                recBuilder.addField(name, nestedVisitorArg.first);
+            }
+        }
+        recBuilder.write(output, true);
+    }
+
+    private void quickSort(int[] index, List<IVisitablePointable> names, int start, int end) {
+        if (end <= start)
+            return;
+        int i = partition(index, names, start, end);
+        quickSort(index, names, start, i - 1);
+        quickSort(index, names, i + 1, end);
+    }
+
+    private int partition(int[] index, List<IVisitablePointable> names, int left, int right) {
+        int i = left - 1;
+        int j = right;
+        while (true) {
+            // grow from the left
+            while (compare(names.get(index[++i]), names.get(index[right])) < 0)
+                ;
+            // lower from the right
+            while (compare(names.get(index[right]), names.get(index[--j])) < 0)
+                if (j == left)
+                    break;
+            if (i >= j)
+                break;
+            // swap i and j
+            swap(index, i, j);
+        }
+        // swap i and right
+        swap(index, i, right); // swap with partition element
+        return i;
+    }
+
+    private void swap(int[] array, int i, int j) {
+        int temp = array[i];
+        array[i] = array[j];
+        array[j] = temp;
+    }
+
+    private int compare(IValueReference a, IValueReference b) {
+        // start+1 and len-1 due to the type tag
+        return fieldNameComparator.compare(a.getByteArray(), a.getStartOffset() + 1, a.getLength() - 1,
+                b.getByteArray(), b.getStartOffset() + 1, b.getLength() - 1);
+    }
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/printer/AListPrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/printer/AListPrinter.java
new file mode 100644
index 0000000..5f47f4f
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/printer/AListPrinter.java
@@ -0,0 +1,85 @@
+/*
+ * 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.om.pointables.printer;
+
+import java.io.IOException;
+import java.io.PrintStream;
+import java.util.List;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.om.pointables.AListPointable;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.EnumDeserializer;
+import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
+
+/**
+ * This class is to print the content of a list. It is ONLY visible to
+ * APrintVisitor.
+ */
+class AListPrinter {
+    private static String LEFT_PAREN = "{{ ";
+    private static String RIGHT_PAREN = " }}";
+    private static String LEFT_PAREN_ORDERED = "[ ";
+    private static String RIGHT_PAREN_ORDERED = " ]";
+    private static String COMMA = ", ";
+
+    private final Pair<PrintStream, ATypeTag> itemVisitorArg = new Pair<PrintStream, ATypeTag>(null, null);
+    private String leftParen = LEFT_PAREN;
+    private String rightParen = RIGHT_PAREN;
+
+    public AListPrinter(boolean ordered) {
+        if (ordered) {
+            leftParen = LEFT_PAREN_ORDERED;
+            rightParen = RIGHT_PAREN_ORDERED;
+        }
+    }
+
+    public void printList(AListPointable listAccessor, PrintStream ps, APrintVisitor visitor) throws IOException,
+            AsterixException {
+        List<IVisitablePointable> itemTags = listAccessor.getItemTags();
+        List<IVisitablePointable> items = listAccessor.getItems();
+        itemVisitorArg.first = ps;
+
+        //print the beginning part
+        ps.print(leftParen);
+
+        // print item 0 to n-2
+        for (int i = 0; i < items.size() - 1; i++) {
+            IVisitablePointable itemTypeTag = itemTags.get(i);
+            IVisitablePointable item = items.get(i);
+            ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(itemTypeTag.getByteArray()[itemTypeTag
+                    .getStartOffset()]);
+            itemVisitorArg.second = item.getLength() <= 1 ? ATypeTag.NULL : typeTag;
+            item.accept(visitor, itemVisitorArg);
+            //print the comma
+            ps.print(COMMA);
+        }
+
+        // print item n-1
+        if (items.size() > 0) {
+            IVisitablePointable itemTypeTag = itemTags.get(itemTags.size() - 1);
+            IVisitablePointable item = items.get(items.size() - 1);
+            ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(itemTypeTag.getByteArray()[itemTypeTag
+                    .getStartOffset()]);
+            itemVisitorArg.second = item.getLength() <= 1 ? ATypeTag.NULL : typeTag;
+            item.accept(visitor, itemVisitorArg);
+        }
+
+        //print the end part
+        ps.print(rightParen);
+    }
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/printer/APrintVisitor.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/printer/APrintVisitor.java
new file mode 100644
index 0000000..3664804
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/printer/APrintVisitor.java
@@ -0,0 +1,185 @@
+/*
+ * 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.om.pointables.printer;
+
+import java.io.PrintStream;
+import java.util.HashMap;
+import java.util.Map;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.ABooleanPrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.ACirclePrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.ADatePrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.ADateTimePrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.ADoublePrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.ADurationPrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.AFloatPrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.AInt16Printer;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.AInt32Printer;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.AInt64Printer;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.AInt8Printer;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.ALinePrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.ANullPrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.APoint3DPrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.APointPrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.APolygonPrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.ARectanglePrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.AStringPrinter;
+import edu.uci.ics.asterix.dataflow.data.nontagged.printers.ATimePrinter;
+import edu.uci.ics.asterix.om.pointables.AFlatValuePointable;
+import edu.uci.ics.asterix.om.pointables.AListPointable;
+import edu.uci.ics.asterix.om.pointables.ARecordPointable;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.pointables.visitor.IVisitablePointableVisitor;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.NotImplementedException;
+import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
+
+/**
+ * This class is a IVisitablePointableVisitor implementation which recursively
+ * visit a given record, list or flat value of a given type, and print it to a
+ * PrintStream in adm format.
+ */
+public class APrintVisitor implements IVisitablePointableVisitor<Void, Pair<PrintStream, ATypeTag>> {
+
+    private final Map<IVisitablePointable, ARecordPrinter> raccessorToPrinter = new HashMap<IVisitablePointable, ARecordPrinter>();
+    private final Map<IVisitablePointable, AListPrinter> laccessorToPrinter = new HashMap<IVisitablePointable, AListPrinter>();
+
+    @Override
+    public Void visit(AListPointable accessor, Pair<PrintStream, ATypeTag> arg) throws AsterixException {
+        AListPrinter printer = laccessorToPrinter.get(accessor);
+        if (printer == null) {
+            printer = new AListPrinter(accessor.ordered());
+            laccessorToPrinter.put(accessor, printer);
+        }
+        try {
+            printer.printList(accessor, arg.first, this);
+        } catch (Exception e) {
+            throw new AsterixException(e);
+        }
+        return null;
+    }
+
+    @Override
+    public Void visit(ARecordPointable accessor, Pair<PrintStream, ATypeTag> arg) throws AsterixException {
+        ARecordPrinter printer = raccessorToPrinter.get(accessor);
+        if (printer == null) {
+            printer = new ARecordPrinter();
+            raccessorToPrinter.put(accessor, printer);
+        }
+        try {
+            printer.printRecord(accessor, arg.first, this);
+        } catch (Exception e) {
+            throw new AsterixException(e);
+        }
+        return null;
+    }
+
+    @Override
+    public Void visit(AFlatValuePointable accessor, Pair<PrintStream, ATypeTag> arg) {
+        try {
+            byte[] b = accessor.getByteArray();
+            int s = accessor.getStartOffset();
+            int l = accessor.getLength();
+            PrintStream ps = arg.first;
+            ATypeTag typeTag = arg.second;
+            switch (typeTag) {
+                case INT8: {
+                    AInt8Printer.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case INT16: {
+                    AInt16Printer.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case INT32: {
+                    AInt32Printer.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case INT64: {
+                    AInt64Printer.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case NULL: {
+                    ANullPrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case BOOLEAN: {
+                    ABooleanPrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case FLOAT: {
+                    AFloatPrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case DOUBLE: {
+                    ADoublePrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case DATE: {
+                    ADatePrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case TIME: {
+                    ATimePrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case DATETIME: {
+                    ADateTimePrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case DURATION: {
+                    ADurationPrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case POINT: {
+                    APointPrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case POINT3D: {
+                    APoint3DPrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case LINE: {
+                    ALinePrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case POLYGON: {
+                    APolygonPrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case CIRCLE: {
+                    ACirclePrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case RECTANGLE: {
+                    ARectanglePrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                case STRING: {
+                    AStringPrinter.INSTANCE.print(b, s, l, ps);
+                    break;
+                }
+                default: {
+                    throw new NotImplementedException("No printer for type " + typeTag);
+                }
+            }
+            return null;
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+    }
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/printer/ARecordPrinter.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/printer/ARecordPrinter.java
new file mode 100644
index 0000000..391d2c9
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/printer/ARecordPrinter.java
@@ -0,0 +1,94 @@
+/*
+ * 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.om.pointables.printer;
+
+import java.io.IOException;
+import java.io.PrintStream;
+import java.util.List;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.om.pointables.ARecordPointable;
+import edu.uci.ics.asterix.om.pointables.base.IVisitablePointable;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.om.types.EnumDeserializer;
+import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
+
+/**
+ * This class is to print the content of a record. It is ONLY visible to
+ * APrintVisitor.
+ */
+class ARecordPrinter {
+    private static String LEFT_PAREN = "{ ";
+    private static String RIGHT_PAREN = " }";
+    private static String COMMA = ", ";
+    private static String COLON = ": ";
+
+    private final Pair<PrintStream, ATypeTag> nameVisitorArg = new Pair<PrintStream, ATypeTag>(null, ATypeTag.STRING);
+    private final Pair<PrintStream, ATypeTag> itemVisitorArg = new Pair<PrintStream, ATypeTag>(null, null);
+
+    public ARecordPrinter() {
+
+    }
+
+    public void printRecord(ARecordPointable recordAccessor, PrintStream ps, APrintVisitor visitor) throws IOException,
+            AsterixException {
+        List<IVisitablePointable> fieldNames = recordAccessor.getFieldNames();
+        List<IVisitablePointable> fieldTags = recordAccessor.getFieldTypeTags();
+        List<IVisitablePointable> fieldValues = recordAccessor.getFieldValues();
+
+        nameVisitorArg.first = ps;
+        itemVisitorArg.first = ps;
+
+        // print the beginning part
+        ps.print(LEFT_PAREN);
+
+        // print field 0 to n-2
+        for (int i = 0; i < fieldNames.size() - 1; i++) {
+            IVisitablePointable itemTypeTag = fieldTags.get(i);
+            IVisitablePointable item = fieldValues.get(i);
+            ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(itemTypeTag.getByteArray()[itemTypeTag
+                    .getStartOffset()]);
+            itemVisitorArg.second = item.getLength() <= 1 ? ATypeTag.NULL : typeTag;
+
+            // print field name
+            fieldNames.get(i).accept(visitor, nameVisitorArg);
+            ps.print(COLON);
+            // print field value
+            item.accept(visitor, itemVisitorArg);
+
+            // print the comma
+            ps.print(COMMA);
+        }
+
+        // print field n-1
+        if (fieldValues.size() > 0) {
+            IVisitablePointable itemTypeTag = fieldTags.get(fieldTags.size() - 1);
+            IVisitablePointable item = fieldValues.get(fieldValues.size() - 1);
+            ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(itemTypeTag.getByteArray()[itemTypeTag
+                    .getStartOffset()]);
+            itemVisitorArg.second = item.getLength() <= 1 ? ATypeTag.NULL : typeTag;
+
+            // print field name
+            fieldNames.get(fieldNames.size() - 1).accept(visitor, nameVisitorArg);
+            ps.print(COLON);
+            // print field value
+            item.accept(visitor, itemVisitorArg);
+        }
+
+        // print the end part
+        ps.print(RIGHT_PAREN);
+    }
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/visitor/IVisitablePointableVisitor.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/visitor/IVisitablePointableVisitor.java
new file mode 100644
index 0000000..bcb4b34
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/pointables/visitor/IVisitablePointableVisitor.java
@@ -0,0 +1,35 @@
+/*
+ * 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.om.pointables.visitor;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.om.pointables.AFlatValuePointable;
+import edu.uci.ics.asterix.om.pointables.AListPointable;
+import edu.uci.ics.asterix.om.pointables.ARecordPointable;
+
+/**
+ * This interface is a visitor for all the three different IVisitablePointable
+ * (Note that right now we have three pointable implementations for type
+ * casting) implementations.
+ */
+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-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/NonTaggedFieldAccessByNameResultType.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/NonTaggedFieldAccessByNameResultType.java
index 4b8b32a9..d9420cf 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/NonTaggedFieldAccessByNameResultType.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/typecomputer/impl/NonTaggedFieldAccessByNameResultType.java
@@ -2,6 +2,7 @@
 
 import edu.uci.ics.asterix.om.base.AString;
 import edu.uci.ics.asterix.om.constants.AsterixConstantValue;
+import edu.uci.ics.asterix.om.pointables.base.DefaultOpenFieldType;
 import edu.uci.ics.asterix.om.typecomputer.base.IResultTypeComputer;
 import edu.uci.ics.asterix.om.types.ARecordType;
 import edu.uci.ics.asterix.om.types.ATypeTag;
@@ -9,7 +10,6 @@
 import edu.uci.ics.asterix.om.types.BuiltinType;
 import edu.uci.ics.asterix.om.types.IAType;
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
-import edu.uci.ics.hyracks.algebricks.common.exceptions.NotImplementedException;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
 import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalExpressionTag;
 import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
@@ -58,7 +58,7 @@
                 return (ARecordType) type0;
             }
             case ANY: {
-                throw new NotImplementedException();
+                return DefaultOpenFieldType.NESTED_OPEN_RECORD_TYPE;
             }
             case UNION: {
                 AUnionType u = (AUnionType) type0;
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/AsterixRuntimeUtil.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/AsterixRuntimeUtil.java
new file mode 100644
index 0000000..624d7eb
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/AsterixRuntimeUtil.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2009-2011 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.om.util;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import edu.uci.ics.asterix.common.api.AsterixAppContextInfoImpl;
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+
+public class AsterixRuntimeUtil {
+
+    public static Set<String> getNodeControllersOnIP(String ipAddress) throws AsterixException {
+        Map<String, Set<String>> nodeControllerInfo = AsterixAppContextInfoImpl.getNodeControllerMap();
+        Set<String> nodeControllersAtLocation = nodeControllerInfo.get(ipAddress);
+        return nodeControllersAtLocation;
+    }
+
+    public static Set<String> getNodeControllersOnHostName(String hostName) throws UnknownHostException {
+        Map<String, Set<String>> nodeControllerInfo = AsterixAppContextInfoImpl.getNodeControllerMap();
+        String address;
+        address = InetAddress.getByName(hostName).getHostAddress();
+        if (address.equals("127.0.1.1")) {
+            address = "127.0.0.1";
+        }
+        Set<String> nodeControllersAtLocation = nodeControllerInfo.get(address);
+        return nodeControllersAtLocation;
+    }
+
+    public static List<String> getAllNodeControllers() {
+
+        Collection<Set<String>> nodeControllersCollection = AsterixAppContextInfoImpl.getNodeControllerMap().values();
+        List<String> nodeControllers = new ArrayList<String>();
+        for (Set<String> ncCollection : nodeControllersCollection) {
+            nodeControllers.addAll(ncCollection);
+        }
+        return nodeControllers;
+    }
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/ResettableByteArrayOutputStream.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/ResettableByteArrayOutputStream.java
new file mode 100644
index 0000000..e47d417
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/ResettableByteArrayOutputStream.java
@@ -0,0 +1,15 @@
+package edu.uci.ics.asterix.om.util;
+
+import edu.uci.ics.hyracks.data.std.util.ByteArrayAccessibleOutputStream;
+
+/**
+ * This class extends ByteArrayAccessibleOutputStream to allow reset to a given
+ * size.
+ * 
+ */
+public class ResettableByteArrayOutputStream extends ByteArrayAccessibleOutputStream {
+
+    public void reset(int size) {
+        count = size;
+    }
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/container/IObjectFactory.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/container/IObjectFactory.java
new file mode 100644
index 0000000..bf11b76
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/container/IObjectFactory.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.om.util.container;
+
+/**
+ * A factory interface to create objects.
+ */
+public interface IObjectFactory<E, T> {
+
+    /**
+     * create an element of type E
+     * 
+     * @param arg
+     * @return an E element
+     */
+    public E create(T arg);
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/container/IObjectPool.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/container/IObjectPool.java
new file mode 100644
index 0000000..4969cd5
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/container/IObjectPool.java
@@ -0,0 +1,36 @@
+/*
+ * 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.om.util.container;
+
+/**
+ * A reusable object pool interface.
+ */
+public interface IObjectPool<E, T> {
+
+    /**
+     * Give client an E instance
+     * 
+     * @param arg
+     *            the argument to create E
+     * @return an E instance
+     */
+    public E allocate(T arg);
+
+    /**
+     * Mark all instances in the pool as unused
+     */
+    public void reset();
+}
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/container/ListObjectPool.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/container/ListObjectPool.java
new file mode 100644
index 0000000..8d9057c
--- /dev/null
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/container/ListObjectPool.java
@@ -0,0 +1,82 @@
+/*
+ * 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.om.util.container;
+
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.List;
+
+/**
+ * Object pool backed by a list.
+ * 
+ * The argument for creating E instances could be different. This class also
+ * considers arguments in object reusing, e.g., it reuses an E instances ONLY
+ * when the construction argument is "equal".
+ */
+public class ListObjectPool<E, T> implements IObjectPool<E, T> {
+
+    private IObjectFactory<E, T> factory;
+
+    /**
+     * cache of objects
+     */
+    private List<E> pool = new ArrayList<E>();
+
+    /**
+     * 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 BitSet usedBits = new BitSet();
+
+    public ListObjectPool(IObjectFactory<E, T> factory) {
+        this.factory = factory;
+    }
+
+    @Override
+    public E allocate(T arg) {
+        int freeSlot = -1;
+        while (freeSlot + 1 < pool.size()) {
+            freeSlot = usedBits.nextClearBit(freeSlot + 1);
+            if (freeSlot >= pool.size())
+                break;
+
+            // the two cases where an element in the pool is a match
+            if ((arg == null && args.get(freeSlot) == null)
+                    || (arg != null && args.get(freeSlot) != null && arg.equals(args.get(freeSlot)))) {
+                // the element is not used and the arg is the same as
+                // input arg
+                usedBits.set(freeSlot);
+                return pool.get(freeSlot);
+            }
+        }
+
+        // if we do not find a reusable object, allocate a new one
+        E element = factory.create(arg);
+        pool.add(element);
+        args.add(arg);
+        usedBits.set(pool.size() - 1);
+        return element;
+    }
+
+    @Override
+    public void reset() {
+        usedBits.clear();
+    }
+}