diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/common/ILogicalBinaryComparator.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/common/ILogicalBinaryComparator.java
index 982a5a0..78ef12d 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/common/ILogicalBinaryComparator.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/common/ILogicalBinaryComparator.java
@@ -21,7 +21,6 @@
 import org.apache.asterix.om.base.IAObject;
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.data.std.api.IPointable;
 
 public interface ILogicalBinaryComparator {
 
@@ -58,11 +57,11 @@
         }
     }
 
-    Result compare(IPointable left, IPointable right) throws HyracksDataException;
+    Result compare(TaggedValueReference left, TaggedValueReference right) throws HyracksDataException;
 
-    Result compare(IPointable left, IAObject rightConstant);
+    Result compare(TaggedValueReference left, IAObject rightConstant);
 
-    Result compare(IAObject leftConstant, IPointable right);
+    Result compare(IAObject leftConstant, TaggedValueReference right);
 
     Result compare(IAObject leftConstant, IAObject rightConstant);
 }
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/common/ListAccessorUtil.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/common/ListAccessorUtil.java
index ba114c0..5350c0c 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/common/ListAccessorUtil.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/common/ListAccessorUtil.java
@@ -19,10 +19,13 @@
 
 package org.apache.asterix.dataflow.data.common;
 
+import static org.apache.asterix.om.types.ATypeTag.VALUE_TYPE_MAPPING;
+
 import java.io.IOException;
 
 import org.apache.asterix.dataflow.data.nontagged.serde.AOrderedListSerializerDeserializer;
 import org.apache.asterix.dataflow.data.nontagged.serde.AUnorderedListSerializerDeserializer;
+import org.apache.asterix.dataflow.data.nontagged.serde.SerializerDeserializerUtil;
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.asterix.om.types.EnumDeserializer;
 import org.apache.asterix.om.utils.NonTaggedFormatUtil;
@@ -52,6 +55,7 @@
      */
     public static boolean getItem(byte[] listBytes, int start, int itemIndex, ATypeTag listTag, ATypeTag listItemTag,
             IPointable pointable, ArrayBackedValueStorage storage) throws IOException {
+        // TODO(ali): this method should be removed if hashing is fixed to avoid copying the tag
         int itemOffset;
         if (listTag == ATypeTag.MULTISET) {
             itemOffset = AUnorderedListSerializerDeserializer.getItemOffset(listBytes, start, itemIndex);
@@ -78,13 +82,24 @@
         }
     }
 
-    public static int numberOfItems(byte[] listBytes, int start) {
-        if (listBytes[start] == ATypeTag.SERIALIZED_UNORDEREDLIST_TYPE_TAG) {
-            return AUnorderedListSerializerDeserializer.getNumberOfItems(listBytes, start);
-        } else if (listBytes[start] == ATypeTag.SERIALIZED_ORDEREDLIST_TYPE_TAG) {
-            return AOrderedListSerializerDeserializer.getNumberOfItems(listBytes, start);
+    public static void getItemFromList(TaggedValueReference listValue, int itemIndex, TaggedValueReference item,
+            ATypeTag arrayItemTag, boolean itemHasTag) throws IOException {
+        int itemOffset = SerializerDeserializerUtil.getItemOffsetNonTagged(listValue, itemIndex);
+        getItem(listValue, itemOffset, item, arrayItemTag, itemHasTag);
+    }
+
+    private static void getItem(TaggedValueReference listValue, int itemOffset, TaggedValueReference item,
+            ATypeTag listItemTag, boolean itemHasTag) throws IOException {
+        byte[] listBytes = listValue.getByteArray();
+        ATypeTag itemTag;
+        int itemValueOffset = itemOffset;
+        if (itemHasTag) {
+            itemTag = VALUE_TYPE_MAPPING[listBytes[itemOffset]];
+            itemValueOffset = itemOffset + 1;
         } else {
-            throw new IllegalStateException();
+            itemTag = listItemTag;
         }
+        int itemValueLength = NonTaggedFormatUtil.getFieldValueLength(listBytes, itemOffset, itemTag, itemHasTag);
+        item.set(listBytes, itemValueOffset, itemValueLength, itemTag);
     }
 }
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/common/TaggedValueReference.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/common/TaggedValueReference.java
new file mode 100644
index 0000000..db50e5e
--- /dev/null
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/common/TaggedValueReference.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 at
+ *
+ *     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 org.apache.asterix.dataflow.data.common;
+
+import java.util.Objects;
+
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.hyracks.data.std.api.IValueReference;
+
+public final class TaggedValueReference implements IValueReference {
+
+    private byte[] bytes;
+    private int valueStart;
+    private int valueLength;
+    private ATypeTag tag;
+
+    public void set(byte[] bytes, int valueStart, int valueLength, ATypeTag tag) {
+        Objects.requireNonNull(tag);
+        this.bytes = bytes;
+        this.valueStart = valueStart;
+        this.valueLength = valueLength;
+        this.tag = tag;
+    }
+
+    @Override
+    public byte[] getByteArray() {
+        return bytes;
+    }
+
+    @Override
+    public int getStartOffset() {
+        return valueStart;
+    }
+
+    @Override
+    public int getLength() {
+        return valueLength;
+    }
+
+    public ATypeTag getTag() {
+        return tag;
+    }
+}
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/AGenericAscBinaryComparator.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/AGenericAscBinaryComparator.java
index f78c4b3..3d2e1d1 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/AGenericAscBinaryComparator.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/AGenericAscBinaryComparator.java
@@ -18,17 +18,25 @@
  */
 package org.apache.asterix.dataflow.data.nontagged.comparators;
 
+import static org.apache.asterix.om.types.ATypeTag.VALUE_TYPE_MAPPING;
+
+import org.apache.asterix.dataflow.data.common.TaggedValueReference;
 import org.apache.asterix.om.types.IAType;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 
 final class AGenericAscBinaryComparator extends AbstractAGenericBinaryComparator {
 
+    private final TaggedValueReference leftValue = new TaggedValueReference();
+    private final TaggedValueReference rightValue = new TaggedValueReference();
+
     AGenericAscBinaryComparator(IAType leftType, IAType rightType) {
         super(leftType, rightType);
     }
 
     @Override
     public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) throws HyracksDataException {
-        return compare(leftType, b1, s1, l1, rightType, b2, s2, l2);
+        leftValue.set(b1, s1 + 1, l1 - 1, VALUE_TYPE_MAPPING[b1[s1]]);
+        rightValue.set(b2, s2 + 1, l2 - 1, VALUE_TYPE_MAPPING[b2[s2]]);
+        return compare(leftType, leftValue, rightType, rightValue);
     }
 }
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/AGenericDescBinaryComparator.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/AGenericDescBinaryComparator.java
index 3f958bf..199fdc6 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/AGenericDescBinaryComparator.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/AGenericDescBinaryComparator.java
@@ -18,11 +18,17 @@
  */
 package org.apache.asterix.dataflow.data.nontagged.comparators;
 
+import static org.apache.asterix.om.types.ATypeTag.VALUE_TYPE_MAPPING;
+
+import org.apache.asterix.dataflow.data.common.TaggedValueReference;
 import org.apache.asterix.om.types.IAType;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 
 final class AGenericDescBinaryComparator extends AbstractAGenericBinaryComparator {
 
+    private final TaggedValueReference leftValue = new TaggedValueReference();
+    private final TaggedValueReference rightValue = new TaggedValueReference();
+
     // interval asc and desc comparators are not the inverse of each other.
     // thus, we need to specify the interval desc comparator factory for descending comparisons.
     AGenericDescBinaryComparator(IAType leftType, IAType rightType) {
@@ -31,11 +37,13 @@
 
     @Override
     public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) throws HyracksDataException {
-        return -compare(leftType, b1, s1, l1, rightType, b2, s2, l2);
+        leftValue.set(b1, s1 + 1, l1 - 1, VALUE_TYPE_MAPPING[b1[s1]]);
+        rightValue.set(b2, s2 + 1, l2 - 1, VALUE_TYPE_MAPPING[b2[s2]]);
+        return -compare(leftType, leftValue, rightType, rightValue);
     }
 
     @Override
     protected int compareInterval(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
-        return -AIntervalDescPartialBinaryComparatorFactory.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+        return -AIntervalDescPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
     }
 }
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/AbstractAGenericBinaryComparator.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/AbstractAGenericBinaryComparator.java
index b3cd642..2e58cd4 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/AbstractAGenericBinaryComparator.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/AbstractAGenericBinaryComparator.java
@@ -19,24 +19,24 @@
 package org.apache.asterix.dataflow.data.nontagged.comparators;
 
 import static org.apache.asterix.om.util.container.ObjectFactories.RECORD_FACTORY;
-import static org.apache.asterix.om.util.container.ObjectFactories.STORAGE_FACTORY;
-import static org.apache.asterix.om.util.container.ObjectFactories.VOID_FACTORY;
+import static org.apache.asterix.om.util.container.ObjectFactories.VALUE_FACTORY;
 
 import java.io.IOException;
 
 import org.apache.asterix.dataflow.data.common.ListAccessorUtil;
+import org.apache.asterix.dataflow.data.common.TaggedValueReference;
 import org.apache.asterix.dataflow.data.nontagged.serde.ADateSerializerDeserializer;
 import org.apache.asterix.dataflow.data.nontagged.serde.ADateTimeSerializerDeserializer;
 import org.apache.asterix.dataflow.data.nontagged.serde.ADayTimeDurationSerializerDeserializer;
 import org.apache.asterix.dataflow.data.nontagged.serde.ATimeSerializerDeserializer;
 import org.apache.asterix.dataflow.data.nontagged.serde.AYearMonthDurationSerializerDeserializer;
+import org.apache.asterix.dataflow.data.nontagged.serde.SerializerDeserializerUtil;
 import org.apache.asterix.om.pointables.nonvisitor.RecordField;
 import org.apache.asterix.om.pointables.nonvisitor.SortedRecord;
 import org.apache.asterix.om.typecomputer.impl.TypeComputeUtils;
 import org.apache.asterix.om.types.ARecordType;
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.asterix.om.types.AbstractCollectionType;
-import org.apache.asterix.om.types.EnumDeserializer;
 import org.apache.asterix.om.types.IAType;
 import org.apache.asterix.om.types.hierachy.ATypeHierarchy;
 import org.apache.asterix.om.types.hierachy.ATypeHierarchy.Domain;
@@ -44,12 +44,9 @@
 import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.data.std.accessors.RawBinaryComparatorFactory;
-import org.apache.hyracks.data.std.api.IMutableValueStorage;
-import org.apache.hyracks.data.std.api.IPointable;
 import org.apache.hyracks.data.std.primitive.BooleanPointable;
 import org.apache.hyracks.data.std.primitive.ByteArrayPointable;
 import org.apache.hyracks.data.std.primitive.UTF8StringPointable;
-import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
 
 /**
  * This comparator is an ordering comparator. It deals with MISSING, NULL, and incompatible types different than the
@@ -59,9 +56,8 @@
 
     protected final IAType leftType;
     protected final IAType rightType;
-    private final ListObjectPool<IMutableValueStorage, Void> storageAllocator = new ListObjectPool<>(STORAGE_FACTORY);
-    private final ListObjectPool<IPointable, Void> voidPointableAllocator = new ListObjectPool<>(VOID_FACTORY);
     private final ListObjectPool<SortedRecord, ARecordType> recordPool = new ListObjectPool<>(RECORD_FACTORY);
+    private final ListObjectPool<TaggedValueReference, Void> taggedValueAllocator = new ListObjectPool<>(VALUE_FACTORY);
 
     AbstractAGenericBinaryComparator(IAType leftType, IAType rightType) {
         // factory should have already made sure to get the actual type (and no null types)
@@ -69,10 +65,10 @@
         this.rightType = rightType;
     }
 
-    protected final int compare(IAType leftType, byte[] b1, int s1, int l1, IAType rightType, byte[] b2, int s2, int l2)
-            throws HyracksDataException {
-        ATypeTag tag1 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b1[s1]);
-        ATypeTag tag2 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b2[s2]);
+    protected final int compare(IAType leftType, TaggedValueReference leftValue, IAType rightType,
+            TaggedValueReference rightValue) throws HyracksDataException {
+        ATypeTag tag1 = leftValue.getTag();
+        ATypeTag tag2 = rightValue.getTag();
         if (tag1 == null || tag2 == null) {
             throw new IllegalStateException("Could not recognize the type of data.");
         }
@@ -86,56 +82,63 @@
         } else if (tag2 == ATypeTag.NULL) {
             return 1;
         }
+        byte[] b1 = leftValue.getByteArray();
+        int s1 = leftValue.getStartOffset();
+        int l1 = leftValue.getLength();
+        byte[] b2 = rightValue.getByteArray();
+        int s2 = rightValue.getStartOffset();
+        int l2 = rightValue.getLength();
+
         if (ATypeHierarchy.isCompatible(tag1, tag2) && ATypeHierarchy.getTypeDomain(tag1) == Domain.NUMERIC) {
-            return ComparatorUtil.compareNumbers(tag1, b1, s1 + 1, tag2, b2, s2 + 1);
+            return ComparatorUtil.compareNumbers(tag1, b1, s1, tag2, b2, s2);
         }
         // currently only numbers are compatible. if two tags are not compatible, we compare the tags to generate order
         if (tag1 != tag2) {
-            return Byte.compare(b1[s1], b2[s2]);
+            return Byte.compare(tag1.serialize(), tag2.serialize());
         }
 
         switch (tag1) {
             case STRING:
-                return UTF8StringPointable.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+                return UTF8StringPointable.compare(b1, s1, l1, b2, s2, l2);
             case UUID:
-                return AUUIDPartialBinaryComparatorFactory.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+                return AUUIDPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
             case BOOLEAN:
-                return BooleanPointable.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+                return BooleanPointable.compare(b1, s1, l1, b2, s2, l2);
             case TIME:
-                return Integer.compare(ATimeSerializerDeserializer.getChronon(b1, s1 + 1),
-                        ATimeSerializerDeserializer.getChronon(b2, s2 + 1));
+                return Integer.compare(ATimeSerializerDeserializer.getChronon(b1, s1),
+                        ATimeSerializerDeserializer.getChronon(b2, s2));
             case DATE:
-                return Integer.compare(ADateSerializerDeserializer.getChronon(b1, s1 + 1),
-                        ADateSerializerDeserializer.getChronon(b2, s2 + 1));
+                return Integer.compare(ADateSerializerDeserializer.getChronon(b1, s1),
+                        ADateSerializerDeserializer.getChronon(b2, s2));
             case YEARMONTHDURATION:
-                return Integer.compare(AYearMonthDurationSerializerDeserializer.getYearMonth(b1, s1 + 1),
-                        AYearMonthDurationSerializerDeserializer.getYearMonth(b2, s2 + 1));
+                return Integer.compare(AYearMonthDurationSerializerDeserializer.getYearMonth(b1, s1),
+                        AYearMonthDurationSerializerDeserializer.getYearMonth(b2, s2));
             case DATETIME:
-                return Long.compare(ADateTimeSerializerDeserializer.getChronon(b1, s1 + 1),
-                        ADateTimeSerializerDeserializer.getChronon(b2, s2 + 1));
+                return Long.compare(ADateTimeSerializerDeserializer.getChronon(b1, s1),
+                        ADateTimeSerializerDeserializer.getChronon(b2, s2));
             case DAYTIMEDURATION:
-                return Long.compare(ADayTimeDurationSerializerDeserializer.getDayTime(b1, s1 + 1),
-                        ADayTimeDurationSerializerDeserializer.getDayTime(b2, s2 + 1));
+                return Long.compare(ADayTimeDurationSerializerDeserializer.getDayTime(b1, s1),
+                        ADayTimeDurationSerializerDeserializer.getDayTime(b2, s2));
             case RECTANGLE:
-                return ARectanglePartialBinaryComparatorFactory.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+                return ARectanglePartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
             case CIRCLE:
-                return ACirclePartialBinaryComparatorFactory.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+                return ACirclePartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
             case POINT:
-                return APointPartialBinaryComparatorFactory.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+                return APointPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
             case POINT3D:
-                return APoint3DPartialBinaryComparatorFactory.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+                return APoint3DPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
             case LINE:
-                return ALinePartialBinaryComparatorFactory.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+                return ALinePartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
             case POLYGON:
-                return APolygonPartialBinaryComparatorFactory.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+                return APolygonPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
             case DURATION:
-                return ADurationPartialBinaryComparatorFactory.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+                return ADurationPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
             case INTERVAL:
                 return compareInterval(b1, s1, l1, b2, s2, l2);
             case BINARY:
-                return ByteArrayPointable.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+                return ByteArrayPointable.compare(b1, s1, l1, b2, s2, l2);
             case ARRAY:
-                return compareArrays(leftType, b1, s1, rightType, b2, s2);
+                return compareArrays(leftType, leftValue, rightType, rightValue);
             case OBJECT:
                 return compareRecords(leftType, b1, s1, rightType, b2, s2);
             default:
@@ -144,32 +147,31 @@
     }
 
     protected int compareInterval(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
-        return AIntervalAscPartialBinaryComparatorFactory.compare(b1, s1 + 1, l1 - 1, b2, s2 + 1, l2 - 1);
+        return AIntervalAscPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
     }
 
-    private int compareArrays(IAType leftType, byte[] b1, int s1, IAType rightType, byte[] b2, int s2)
-            throws HyracksDataException {
-        int leftNumItems = ListAccessorUtil.numberOfItems(b1, s1);
-        int rightNumItems = ListAccessorUtil.numberOfItems(b2, s2);
-        IAType leftArrayType = TypeComputeUtils.getActualTypeOrOpen(leftType, ATypeTag.ARRAY);
-        IAType rightArrayType = TypeComputeUtils.getActualTypeOrOpen(rightType, ATypeTag.ARRAY);
-        IAType leftItemType = ((AbstractCollectionType) leftArrayType).getItemType();
-        IAType rightItemType = ((AbstractCollectionType) rightArrayType).getItemType();
-        ATypeTag leftItemTag = leftItemType.getTypeTag();
-        ATypeTag rightItemTag = rightItemType.getTypeTag();
-        // TODO(ali): could be optimized to not need pointable when changing comparator to be non-tagged & no visitable
-        IPointable leftItem = voidPointableAllocator.allocate(null);
-        IPointable rightItem = voidPointableAllocator.allocate(null);
-        // TODO(ali): optimize to not need this storage, will require optimizing records comparison to not use visitable
-        ArrayBackedValueStorage leftStorage = (ArrayBackedValueStorage) storageAllocator.allocate(null);
-        ArrayBackedValueStorage rightStorage = (ArrayBackedValueStorage) storageAllocator.allocate(null);
+    private int compareArrays(IAType leftArrayType, TaggedValueReference leftArray, IAType rightArrayType,
+            TaggedValueReference rightArray) throws HyracksDataException {
+        int leftNumItems = SerializerDeserializerUtil.getNumberOfItemsNonTagged(leftArray);
+        int rightNumItems = SerializerDeserializerUtil.getNumberOfItemsNonTagged(rightArray);
+        IAType leftItemType =
+                ((AbstractCollectionType) TypeComputeUtils.getActualTypeOrOpen(leftArrayType, ATypeTag.ARRAY))
+                        .getItemType();
+        IAType rightItemType =
+                ((AbstractCollectionType) TypeComputeUtils.getActualTypeOrOpen(rightArrayType, ATypeTag.ARRAY))
+                        .getItemType();
+        ATypeTag leftArrayItemTag = leftItemType.getTypeTag();
+        ATypeTag rightArrayItemTag = rightItemType.getTypeTag();
+        TaggedValueReference leftItem = taggedValueAllocator.allocate(null);
+        TaggedValueReference rightItem = taggedValueAllocator.allocate(null);
+        boolean leftItemHasTag = leftArrayItemTag == ATypeTag.ANY;
+        boolean rightItemHasTag = rightArrayItemTag == ATypeTag.ANY;
         int result;
         try {
             for (int i = 0; i < leftNumItems && i < rightNumItems; i++) {
-                ListAccessorUtil.getItem(b1, s1, i, ATypeTag.ARRAY, leftItemTag, leftItem, leftStorage);
-                ListAccessorUtil.getItem(b2, s2, i, ATypeTag.ARRAY, rightItemTag, rightItem, rightStorage);
-                result = compare(leftItemType, leftItem.getByteArray(), leftItem.getStartOffset(), leftItem.getLength(),
-                        rightItemType, rightItem.getByteArray(), rightItem.getStartOffset(), rightItem.getLength());
+                ListAccessorUtil.getItemFromList(leftArray, i, leftItem, leftArrayItemTag, leftItemHasTag);
+                ListAccessorUtil.getItemFromList(rightArray, i, rightItem, rightArrayItemTag, rightItemHasTag);
+                result = compare(leftItemType, leftItem, rightItemType, rightItem);
                 if (result != 0) {
                     return result;
                 }
@@ -178,10 +180,8 @@
         } catch (IOException e) {
             throw HyracksDataException.create(e);
         } finally {
-            storageAllocator.free(rightStorage);
-            storageAllocator.free(leftStorage);
-            voidPointableAllocator.free(rightItem);
-            voidPointableAllocator.free(leftItem);
+            taggedValueAllocator.free(rightItem);
+            taggedValueAllocator.free(leftItem);
         }
     }
 
@@ -191,35 +191,25 @@
         ARecordType rightRecordType = (ARecordType) TypeComputeUtils.getActualTypeOrOpen(rightType, ATypeTag.OBJECT);
         SortedRecord leftRecord = recordPool.allocate(leftRecordType);
         SortedRecord rightRecord = recordPool.allocate(rightRecordType);
-        IPointable leftFieldValue = voidPointableAllocator.allocate(null);
-        IPointable rightFieldValue = voidPointableAllocator.allocate(null);
-        // TODO(ali): this is not ideal. should be removed when tagged pointables are introduced
-        ArrayBackedValueStorage leftStorage = (ArrayBackedValueStorage) storageAllocator.allocate(null);
-        ArrayBackedValueStorage rightStorage = (ArrayBackedValueStorage) storageAllocator.allocate(null);
+        TaggedValueReference leftFieldValue = taggedValueAllocator.allocate(null);
+        TaggedValueReference rightFieldValue = taggedValueAllocator.allocate(null);
         try {
-            leftRecord.reset(b1, s1);
-            rightRecord.reset(b2, s2);
-            IAType leftFieldType, rightFieldType;
-            RecordField leftField, rightField;
+            leftRecord.resetNonTagged(b1, s1);
+            rightRecord.resetNonTagged(b2, s2);
             int result;
             while (!leftRecord.isEmpty() && !rightRecord.isEmpty()) {
-                leftField = leftRecord.poll();
-                rightField = rightRecord.poll();
+                RecordField leftField = leftRecord.poll();
+                RecordField rightField = rightRecord.poll();
                 // compare the names first
                 result = RecordField.FIELD_NAME_COMP.compare(leftField, rightField);
                 if (result != 0) {
                     return result;
                 }
                 // then compare the values if the names are equal
-                leftStorage.reset();
-                rightStorage.reset();
-                leftRecord.getFieldValue(leftField, leftFieldValue, leftStorage);
-                rightRecord.getFieldValue(rightField, rightFieldValue, rightStorage);
-                leftFieldType = leftRecord.getFieldType(leftField);
-                rightFieldType = rightRecord.getFieldType(rightField);
-                result = compare(leftFieldType, leftFieldValue.getByteArray(), leftFieldValue.getStartOffset(),
-                        leftFieldValue.getLength(), rightFieldType, rightFieldValue.getByteArray(),
-                        rightFieldValue.getStartOffset(), rightFieldValue.getLength());
+                leftRecord.getFieldValue(leftField, leftFieldValue);
+                rightRecord.getFieldValue(rightField, rightFieldValue);
+                result = compare(leftRecord.getFieldType(leftField), leftFieldValue,
+                        rightRecord.getFieldType(rightField), rightFieldValue);
                 if (result != 0) {
                     return result;
                 }
@@ -230,10 +220,8 @@
         } finally {
             recordPool.free(rightRecord);
             recordPool.free(leftRecord);
-            voidPointableAllocator.free(rightFieldValue);
-            voidPointableAllocator.free(leftFieldValue);
-            storageAllocator.free(rightStorage);
-            storageAllocator.free(leftStorage);
+            taggedValueAllocator.free(rightFieldValue);
+            taggedValueAllocator.free(leftFieldValue);
         }
     }
 }
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/ComparatorUtil.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/ComparatorUtil.java
index 6f08b37..4db23b2 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/ComparatorUtil.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/ComparatorUtil.java
@@ -30,6 +30,7 @@
 
 import org.apache.asterix.dataflow.data.common.ILogicalBinaryComparator;
 import org.apache.asterix.dataflow.data.common.ILogicalBinaryComparator.Result;
+import org.apache.asterix.dataflow.data.common.TaggedValueReference;
 import org.apache.asterix.dataflow.data.nontagged.serde.ADoubleSerializerDeserializer;
 import org.apache.asterix.dataflow.data.nontagged.serde.AFloatSerializerDeserializer;
 import org.apache.asterix.dataflow.data.nontagged.serde.AInt16SerializerDeserializer;
@@ -47,7 +48,6 @@
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.asterix.om.types.IAType;
 import org.apache.asterix.om.types.hierachy.ATypeHierarchy;
-import org.apache.hyracks.data.std.api.IPointable;
 
 // TODO(ali): refactor some functionality with ATypeHierarchy and others
 public class ComparatorUtil {
@@ -81,13 +81,7 @@
         return null;
     }
 
-    // checking that left and right are compatible and are numbers has to be done before calling this
-    static Result compareNumbers(ATypeTag leftTag, IPointable left, ATypeTag rightTag, IPointable right) {
-        return asResult(compareNumbers(leftTag, left.getByteArray(), left.getStartOffset() + 1, rightTag,
-                right.getByteArray(), right.getStartOffset() + 1));
-    }
-
-    // start args point to the value
+    // start points to the value; checking left and right are compatible and numbers has to be done before calling this
     static int compareNumbers(ATypeTag lTag, byte[] l, int lStart, ATypeTag rTag, byte[] r, int rStart) {
         if (lTag == DOUBLE || rTag == DOUBLE) {
             return Double.compare(getDoubleValue(lTag, l, lStart), getDoubleValue(rTag, r, rStart));
@@ -103,10 +97,11 @@
     }
 
     // checking that left and right are compatible has to be done before calling this
-    static Result compareNumWithConstant(ATypeTag leftTag, IPointable left, IAObject right) {
+    static Result compareNumWithConstant(TaggedValueReference left, IAObject right) {
+        ATypeTag leftTag = left.getTag();
         ATypeTag rightTag = right.getType().getTypeTag();
         byte[] leftBytes = left.getByteArray();
-        int start = left.getStartOffset() + 1;
+        int start = left.getStartOffset();
         if (leftTag == DOUBLE || rightTag == DOUBLE) {
             return asResult(Double.compare(getDoubleValue(leftTag, leftBytes, start), getConstantDouble(right)));
         } else if (leftTag == FLOAT || rightTag == FLOAT) {
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/LogicalComplexBinaryComparator.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/LogicalComplexBinaryComparator.java
index a4fce6d..93ccaa3 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/LogicalComplexBinaryComparator.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/LogicalComplexBinaryComparator.java
@@ -18,38 +18,27 @@
  */
 package org.apache.asterix.dataflow.data.nontagged.comparators;
 
-import static org.apache.asterix.om.types.ATypeTag.SERIALIZED_MISSING_TYPE_TAG;
-import static org.apache.asterix.om.types.ATypeTag.VALUE_TYPE_MAPPING;
-import static org.apache.asterix.om.util.container.ObjectFactories.BIT_SET_FACTORY;
-import static org.apache.asterix.om.util.container.ObjectFactories.STORAGE_FACTORY;
-import static org.apache.asterix.om.util.container.ObjectFactories.VOID_FACTORY;
+import static org.apache.asterix.om.util.container.ObjectFactories.RECORD_FACTORY;
+import static org.apache.asterix.om.util.container.ObjectFactories.VALUE_FACTORY;
 
 import java.io.IOException;
-import java.util.BitSet;
-import java.util.List;
 
 import org.apache.asterix.dataflow.data.common.ILogicalBinaryComparator;
 import org.apache.asterix.dataflow.data.common.ListAccessorUtil;
+import org.apache.asterix.dataflow.data.common.TaggedValueReference;
+import org.apache.asterix.dataflow.data.nontagged.serde.SerializerDeserializerUtil;
 import org.apache.asterix.om.base.IAObject;
-import org.apache.asterix.om.pointables.ARecordVisitablePointable;
-import org.apache.asterix.om.pointables.PointableAllocator;
-import org.apache.asterix.om.pointables.base.IVisitablePointable;
+import org.apache.asterix.om.pointables.nonvisitor.RecordField;
+import org.apache.asterix.om.pointables.nonvisitor.SortedRecord;
 import org.apache.asterix.om.typecomputer.impl.TypeComputeUtils;
 import org.apache.asterix.om.types.ARecordType;
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.asterix.om.types.AbstractCollectionType;
-import org.apache.asterix.om.types.EnumDeserializer;
 import org.apache.asterix.om.types.IAType;
-import org.apache.asterix.om.util.container.IObjectPool;
 import org.apache.asterix.om.util.container.ListObjectPool;
 import org.apache.asterix.om.utils.RecordUtil;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.data.std.accessors.RawBinaryComparatorFactory;
-import org.apache.hyracks.data.std.api.IMutableValueStorage;
-import org.apache.hyracks.data.std.api.IPointable;
-import org.apache.hyracks.data.std.api.IValueReference;
-import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
-import org.apache.hyracks.util.string.UTF8StringUtil;
 
 public final class LogicalComplexBinaryComparator implements ILogicalBinaryComparator {
 
@@ -57,26 +46,20 @@
     private final IAType rightType;
     private final boolean isEquality;
     private final LogicalScalarBinaryComparator scalarComparator;
-    private final IObjectPool<IMutableValueStorage, Void> storageAllocator;
-    private final IObjectPool<IPointable, Void> voidPointableAllocator;
-    private final IObjectPool<BitSet, Void> bitSetAllocator;
-    private final PointableAllocator pointableAllocator;
+    private final ListObjectPool<TaggedValueReference, Void> taggedValuePool = new ListObjectPool<>(VALUE_FACTORY);
+    private final ListObjectPool<SortedRecord, ARecordType> recordPool = new ListObjectPool<>(RECORD_FACTORY);
 
     LogicalComplexBinaryComparator(IAType leftType, IAType rightType, boolean isEquality) {
         this.leftType = leftType;
         this.rightType = rightType;
         this.isEquality = isEquality;
         this.scalarComparator = LogicalScalarBinaryComparator.of(isEquality);
-        storageAllocator = new ListObjectPool<>(STORAGE_FACTORY);
-        voidPointableAllocator = new ListObjectPool<>(VOID_FACTORY);
-        bitSetAllocator = new ListObjectPool<>(BIT_SET_FACTORY);
-        pointableAllocator = new PointableAllocator();
     }
 
     @Override
-    public Result compare(IPointable left, IPointable right) throws HyracksDataException {
-        ATypeTag leftRuntimeTag = VALUE_TYPE_MAPPING[left.getByteArray()[left.getStartOffset()]];
-        ATypeTag rightRuntimeTag = VALUE_TYPE_MAPPING[right.getByteArray()[right.getStartOffset()]];
+    public Result compare(TaggedValueReference left, TaggedValueReference right) throws HyracksDataException {
+        ATypeTag leftRuntimeTag = left.getTag();
+        ATypeTag rightRuntimeTag = right.getTag();
         Result comparisonResult = ComparatorUtil.returnMissingOrNullOrMismatch(leftRuntimeTag, rightRuntimeTag);
         if (comparisonResult != null) {
             return comparisonResult;
@@ -89,9 +72,9 @@
     }
 
     @Override
-    public Result compare(IPointable left, IAObject rightConstant) {
+    public Result compare(TaggedValueReference left, IAObject rightConstant) {
         // TODO(ali): not defined currently for constant complex types
-        ATypeTag leftTag = VALUE_TYPE_MAPPING[left.getByteArray()[left.getStartOffset()]];
+        ATypeTag leftTag = left.getTag();
         ATypeTag rightTag = rightConstant.getType().getTypeTag();
         Result comparisonResult = ComparatorUtil.returnMissingOrNullOrMismatch(leftTag, rightTag);
         if (comparisonResult != null) {
@@ -102,7 +85,7 @@
     }
 
     @Override
-    public Result compare(IAObject leftConstant, IPointable right) {
+    public Result compare(IAObject leftConstant, TaggedValueReference right) {
         // TODO(ali): not defined currently for constant complex types
         Result result = compare(right, leftConstant);
         switch (result) {
@@ -128,18 +111,19 @@
         return Result.NULL;
     }
 
-    private Result compareComplex(IAType leftType, ATypeTag leftTag, IPointable left, IAType rightType,
-            ATypeTag rightTag, IPointable right) throws HyracksDataException {
-        if (leftTag != rightTag) {
+    private Result compareComplex(IAType leftType, ATypeTag leftRuntimeTag, TaggedValueReference left, IAType rightType,
+            ATypeTag rightRuntimeTag, TaggedValueReference right) throws HyracksDataException {
+        if (leftRuntimeTag != rightRuntimeTag) {
             return Result.INCOMPARABLE;
         }
-        IAType leftCompileType = TypeComputeUtils.getActualTypeOrOpen(leftType, leftTag);
-        IAType rightCompileType = TypeComputeUtils.getActualTypeOrOpen(rightType, rightTag);
-        switch (leftTag) {
+        IAType leftCompileType = TypeComputeUtils.getActualTypeOrOpen(leftType, leftRuntimeTag);
+        IAType rightCompileType = TypeComputeUtils.getActualTypeOrOpen(rightType, rightRuntimeTag);
+        switch (leftRuntimeTag) {
             case MULTISET:
-                return compareMultisets(leftCompileType, leftTag, left, rightCompileType, rightTag, right);
+                return compareMultisets(leftCompileType, leftRuntimeTag, left, rightCompileType, rightRuntimeTag,
+                        right);
             case ARRAY:
-                return compareArrays(leftCompileType, leftTag, left, rightCompileType, rightTag, right);
+                return compareArrays(leftCompileType, left, rightCompileType, right);
             case OBJECT:
                 return compareRecords(leftCompileType, left, rightCompileType, right);
             default:
@@ -147,41 +131,28 @@
         }
     }
 
-    private Result compareArrays(IAType leftType, ATypeTag leftListTag, IPointable left, IAType rightType,
-            ATypeTag rightListTag, IPointable right) throws HyracksDataException {
-        // reaching here, both left and right have to be arrays (should be enforced)
-        byte[] leftBytes = left.getByteArray();
-        byte[] rightBytes = right.getByteArray();
-        int leftStart = left.getStartOffset();
-        int rightStart = right.getStartOffset();
-        int leftNumItems = ListAccessorUtil.numberOfItems(leftBytes, leftStart);
-        int rightNumItems = ListAccessorUtil.numberOfItems(rightBytes, rightStart);
+    private Result compareArrays(IAType leftType, TaggedValueReference leftArray, IAType rightType,
+            TaggedValueReference rightArray) throws HyracksDataException {
+        // reaching here, both leftArray and rightArray have to be arrays (should be enforced)
+        int leftNumItems = SerializerDeserializerUtil.getNumberOfItemsNonTagged(leftArray);
+        int rightNumItems = SerializerDeserializerUtil.getNumberOfItemsNonTagged(rightArray);
         IAType leftItemCompileType = ((AbstractCollectionType) leftType).getItemType();
         IAType rightItemCompileType = ((AbstractCollectionType) rightType).getItemType();
-        ATypeTag leftItemTag = leftItemCompileType.getTypeTag();
-        ATypeTag rightItemTag = rightItemCompileType.getTypeTag();
-
-        // TODO(ali): could be optimized to not need pointable when changing comparator to be non-tagged & no visitable
-        IPointable leftItem = voidPointableAllocator.allocate(null);
-        IPointable rightItem = voidPointableAllocator.allocate(null);
-        // TODO(ali): optimize to not need this storage, will require optimizing records comparison to not use visitable
-        ArrayBackedValueStorage leftStorage = (ArrayBackedValueStorage) storageAllocator.allocate(null);
-        ArrayBackedValueStorage rightStorage = (ArrayBackedValueStorage) storageAllocator.allocate(null);
+        ATypeTag leftArrayItemTag = leftItemCompileType.getTypeTag();
+        ATypeTag rightArrayItemTag = rightItemCompileType.getTypeTag();
+        boolean leftItemHasTag = leftArrayItemTag == ATypeTag.ANY;
+        boolean rightItemHasTag = rightArrayItemTag == ATypeTag.ANY;
+        TaggedValueReference leftItem = taggedValuePool.allocate(null);
+        TaggedValueReference rightItem = taggedValuePool.allocate(null);
         Result determiningResult = null;
         Result tempResult;
-        byte leftItemTagByte, rightItemTagByte;
-        ATypeTag leftItemRuntimeTag, rightItemRuntimeTag;
         try {
             for (int i = 0; i < leftNumItems && i < rightNumItems; i++) {
-                ListAccessorUtil.getItem(leftBytes, leftStart, i, leftListTag, leftItemTag, leftItem, leftStorage);
-                ListAccessorUtil.getItem(rightBytes, rightStart, i, rightListTag, rightItemTag, rightItem,
-                        rightStorage);
-                leftItemTagByte = leftItem.getByteArray()[leftItem.getStartOffset()];
-                rightItemTagByte = rightItem.getByteArray()[rightItem.getStartOffset()];
-
+                ListAccessorUtil.getItemFromList(leftArray, i, leftItem, leftArrayItemTag, leftItemHasTag);
+                ListAccessorUtil.getItemFromList(rightArray, i, rightItem, rightArrayItemTag, rightItemHasTag);
                 // if both tags are derived, get item type or default to open item if array is open, then call complex
-                leftItemRuntimeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(leftItemTagByte);
-                rightItemRuntimeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(rightItemTagByte);
+                ATypeTag leftItemRuntimeTag = leftItem.getTag();
+                ATypeTag rightItemRuntimeTag = rightItem.getTag();
                 if (leftItemRuntimeTag.isDerivedType() && rightItemRuntimeTag.isDerivedType()) {
                     tempResult = compareComplex(leftItemCompileType, leftItemRuntimeTag, leftItem, rightItemCompileType,
                             rightItemRuntimeTag, rightItem);
@@ -206,15 +177,13 @@
         } catch (IOException e) {
             throw HyracksDataException.create(e);
         } finally {
-            storageAllocator.free(rightStorage);
-            storageAllocator.free(leftStorage);
-            voidPointableAllocator.free(rightItem);
-            voidPointableAllocator.free(leftItem);
+            taggedValuePool.free(rightItem);
+            taggedValuePool.free(leftItem);
         }
     }
 
-    private Result compareMultisets(IAType leftType, ATypeTag leftListTag, IPointable left, IAType rightType,
-            ATypeTag rightListTag, IPointable right) throws HyracksDataException {
+    private Result compareMultisets(IAType leftType, ATypeTag leftListTag, TaggedValueReference left, IAType rightType,
+            ATypeTag rightListTag, TaggedValueReference right) throws HyracksDataException {
         // TODO(ali): multiset comparison logic here
         // equality is the only operation defined for multiset
         if (!isEquality) {
@@ -225,107 +194,78 @@
                         left.getLength(), right.getByteArray(), right.getStartOffset(), right.getLength()));
     }
 
-    private Result compareRecords(IAType leftType, IPointable left, IAType rightType, IPointable right)
-            throws HyracksDataException {
+    private Result compareRecords(IAType leftType, TaggedValueReference left, IAType rightType,
+            TaggedValueReference right) throws HyracksDataException {
         // equality is the only operation defined for records
         if (!isEquality) {
             return Result.INCOMPARABLE;
         }
-        ARecordType leftRecordType = (ARecordType) leftType;
-        ARecordType rightRecordType = (ARecordType) rightType;
-        ARecordVisitablePointable leftRecord = pointableAllocator.allocateRecordValue(leftRecordType);
-        ARecordVisitablePointable rightRecord = pointableAllocator.allocateRecordValue(rightRecordType);
-        // keeps track of the fields in the right record that have not been matched
-        BitSet notMatched = bitSetAllocator.allocate(null);
+        ARecordType leftRecordType = (ARecordType) TypeComputeUtils.getActualTypeOrOpen(leftType, ATypeTag.OBJECT);
+        ARecordType rightRecordType = (ARecordType) TypeComputeUtils.getActualTypeOrOpen(rightType, ATypeTag.OBJECT);
+        SortedRecord leftRecord = recordPool.allocate(leftRecordType);
+        SortedRecord rightRecord = recordPool.allocate(rightRecordType);
+        TaggedValueReference leftFieldValue = taggedValuePool.allocate(null);
+        TaggedValueReference rightFieldValue = taggedValuePool.allocate(null);
         try {
-            leftRecord.set(left);
-            rightRecord.set(right);
-            List<IVisitablePointable> leftFieldValues = leftRecord.getFieldValues();
-            List<IVisitablePointable> leftFieldNames = leftRecord.getFieldNames();
-            List<IVisitablePointable> rightFieldValues = rightRecord.getFieldValues();
-            List<IVisitablePointable> rightFieldNames = rightRecord.getFieldNames();
-            IVisitablePointable leftFieldValue, leftFieldName, rightFieldValue, rightFieldName;
-            int leftNumFields = leftFieldNames.size();
-            int rightNumFields = rightFieldNames.size();
-            IAType leftFieldType, rightFieldType;
-            ATypeTag leftFTag, rightFTag;
-            Result tempCompResult;
-            boolean foundFieldInRight;
+            leftRecord.resetNonTagged(left.getByteArray(), left.getStartOffset());
+            rightRecord.resetNonTagged(right.getByteArray(), right.getStartOffset());
             boolean notEqual = false;
-            notMatched.set(0, rightNumFields);
-            for (int i = 0; i < leftNumFields; i++) {
-                leftFieldValue = leftFieldValues.get(i);
-                leftFTag = VALUE_TYPE_MAPPING[leftFieldValue.getByteArray()[leftFieldValue.getStartOffset()]];
-
-                // ignore if the field value is missing
-                if (leftFTag != ATypeTag.MISSING) {
-                    // start looking for the field in the right record
-                    foundFieldInRight = false;
-                    leftFieldName = leftFieldNames.get(i);
-                    for (int k = 0; k < rightNumFields; k++) {
-                        rightFieldName = rightFieldNames.get(k);
-                        if (notMatched.get(k) && equalNames(leftFieldName, rightFieldName)) {
-                            notMatched.clear(k);
-                            rightFieldValue = rightFieldValues.get(k);
-                            rightFTag = VALUE_TYPE_MAPPING[rightFieldValue.getByteArray()[rightFieldValue
-                                    .getStartOffset()]];
-                            // if right field has a missing value, ignore and flag the two records as not equal
-                            if (rightFTag != ATypeTag.MISSING) {
-                                foundFieldInRight = true;
-                                if (leftFTag == ATypeTag.NULL || rightFTag == ATypeTag.NULL) {
-                                    tempCompResult = Result.NULL;
-                                } else if (leftFTag.isDerivedType() && rightFTag.isDerivedType()) {
-                                    leftFieldType = RecordUtil.getType(leftRecordType, i, leftFTag);
-                                    rightFieldType = RecordUtil.getType(rightRecordType, k, rightFTag);
-                                    tempCompResult = compareComplex(leftFieldType, leftFTag, leftFieldValue,
-                                            rightFieldType, rightFTag, rightFieldValue);
-                                } else {
-                                    tempCompResult = scalarComparator.compare(leftFieldValue, rightFieldValue);
-                                }
-
-                                if (tempCompResult == Result.INCOMPARABLE || tempCompResult == Result.MISSING
-                                        || tempCompResult == Result.NULL) {
-                                    return Result.INCOMPARABLE;
-                                }
-                                if (tempCompResult != Result.EQ) {
-                                    notEqual = true;
-                                }
-                            }
-                            break;
-                        }
+            RecordField leftField = null, rightField = null;
+            int previousNamesComparisonResult = 0;
+            while (!leftRecord.isEmpty() && !rightRecord.isEmpty()) {
+                if (previousNamesComparisonResult == 0) {
+                    // previous field names were equal or first time to enter the loop
+                    leftField = leftRecord.poll();
+                    rightField = rightRecord.poll();
+                } else if (previousNamesComparisonResult > 0) {
+                    // right field name was less than left field name. get next field from right
+                    rightField = rightRecord.poll();
+                } else {
+                    leftField = leftRecord.poll();
+                }
+                Result tempCompResult;
+                previousNamesComparisonResult = RecordField.FIELD_NAME_COMP.compare(leftField, rightField);
+                if (previousNamesComparisonResult == 0) {
+                    // filed names are equal
+                    leftRecord.getFieldValue(leftField, leftFieldValue);
+                    rightRecord.getFieldValue(rightField, rightFieldValue);
+                    ATypeTag leftFTag = leftFieldValue.getTag();
+                    ATypeTag rightFTag = rightFieldValue.getTag();
+                    if (leftFTag == ATypeTag.NULL || rightFTag == ATypeTag.NULL) {
+                        tempCompResult = Result.NULL;
+                    } else if (leftFTag.isDerivedType() && rightFTag.isDerivedType()) {
+                        IAType leftFieldType = RecordUtil.getType(leftRecordType, leftField.getIndex(), leftFTag);
+                        IAType rightFieldType = RecordUtil.getType(rightRecordType, rightField.getIndex(), rightFTag);
+                        tempCompResult = compareComplex(leftFieldType, leftFTag, leftFieldValue, rightFieldType,
+                                rightFTag, rightFieldValue);
+                    } else {
+                        tempCompResult = scalarComparator.compare(leftFieldValue, rightFieldValue);
                     }
-                    if (!foundFieldInRight) {
+
+                    if (tempCompResult == Result.INCOMPARABLE || tempCompResult == Result.MISSING
+                            || tempCompResult == Result.NULL) {
+                        return Result.INCOMPARABLE;
+                    }
+                    if (tempCompResult != Result.EQ) {
                         notEqual = true;
                     }
+                } else {
+                    notEqual = true;
                 }
             }
-
-            if (notEqual) {
+            if (notEqual || leftRecord.size() != rightRecord.size()) {
                 // LT or GT does not make a difference since this is an answer to equality
                 return Result.LT;
             }
-            // check if there is a field in the right record that does not exist in left record
-            byte rightFieldTag;
-            for (int i = notMatched.nextSetBit(0); i >= 0 && i < rightNumFields; i = notMatched.nextSetBit(i + 1)) {
-                rightFieldValue = rightFieldValues.get(i);
-                rightFieldTag = rightFieldValue.getByteArray()[rightFieldValue.getStartOffset()];
-                if (rightFieldTag != SERIALIZED_MISSING_TYPE_TAG) {
-                    // LT or GT does not make a difference since this is an answer to equality
-                    return Result.LT;
-                }
-            }
-
             // reaching here means every field in the left record exists in the right and vice versa
             return Result.EQ;
+        } catch (IOException e) {
+            throw HyracksDataException.create(e);
         } finally {
-            pointableAllocator.freeRecord(rightRecord);
-            pointableAllocator.freeRecord(leftRecord);
-            bitSetAllocator.free(notMatched);
+            recordPool.free(rightRecord);
+            recordPool.free(leftRecord);
+            taggedValuePool.free(rightFieldValue);
+            taggedValuePool.free(leftFieldValue);
         }
     }
-
-    private boolean equalNames(IValueReference fieldName1, IValueReference fieldName2) {
-        return UTF8StringUtil.compareTo(fieldName1.getByteArray(), fieldName1.getStartOffset() + 1,
-                fieldName2.getByteArray(), fieldName2.getStartOffset() + 1) == 0;
-    }
 }
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/LogicalGenericBinaryComparator.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/LogicalGenericBinaryComparator.java
index 57dbb33..d3c4183 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/LogicalGenericBinaryComparator.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/LogicalGenericBinaryComparator.java
@@ -18,14 +18,11 @@
  */
 package org.apache.asterix.dataflow.data.nontagged.comparators;
 
-import static org.apache.asterix.om.types.ATypeTag.VALUE_TYPE_MAPPING;
-
 import org.apache.asterix.dataflow.data.common.ILogicalBinaryComparator;
+import org.apache.asterix.dataflow.data.common.TaggedValueReference;
 import org.apache.asterix.om.base.IAObject;
-import org.apache.asterix.om.types.ATypeTag;
 import org.apache.asterix.om.types.IAType;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.data.std.api.IPointable;
 
 public final class LogicalGenericBinaryComparator implements ILogicalBinaryComparator {
 
@@ -38,27 +35,23 @@
     }
 
     @Override
-    public Result compare(IPointable left, IPointable right) throws HyracksDataException {
-        ATypeTag leftTag = VALUE_TYPE_MAPPING[left.getByteArray()[left.getStartOffset()]];
-        ATypeTag rightTag = VALUE_TYPE_MAPPING[right.getByteArray()[right.getStartOffset()]];
-        if (leftTag.isDerivedType() && rightTag.isDerivedType()) {
+    public Result compare(TaggedValueReference left, TaggedValueReference right) throws HyracksDataException {
+        if (left.getTag().isDerivedType() && right.getTag().isDerivedType()) {
             return complexComparator.compare(left, right);
         }
         return scalarComparator.compare(left, right);
     }
 
     @Override
-    public Result compare(IPointable left, IAObject rightConstant) {
-        ATypeTag leftTag = VALUE_TYPE_MAPPING[left.getByteArray()[left.getStartOffset()]];
-        ATypeTag rightTag = rightConstant.getType().getTypeTag();
-        if (leftTag.isDerivedType() && rightTag.isDerivedType()) {
+    public Result compare(TaggedValueReference left, IAObject rightConstant) {
+        if (left.getTag().isDerivedType() && rightConstant.getType().getTypeTag().isDerivedType()) {
             return complexComparator.compare(left, rightConstant);
         }
         return scalarComparator.compare(left, rightConstant);
     }
 
     @Override
-    public Result compare(IAObject leftConstant, IPointable right) {
+    public Result compare(IAObject leftConstant, TaggedValueReference right) {
         Result result = compare(right, leftConstant);
         if (result == Result.LT) {
             return Result.GT;
@@ -69,12 +62,10 @@
     }
 
     @Override
-    public Result compare(IAObject leftConstant, IAObject rightConstant) {
-        ATypeTag leftTag = leftConstant.getType().getTypeTag();
-        ATypeTag rightTag = rightConstant.getType().getTypeTag();
-        if (leftTag.isDerivedType() && rightTag.isDerivedType()) {
-            return complexComparator.compare(leftConstant, rightConstant);
+    public Result compare(IAObject leftConst, IAObject rightConst) {
+        if (leftConst.getType().getTypeTag().isDerivedType() && rightConst.getType().getTypeTag().isDerivedType()) {
+            return complexComparator.compare(leftConst, rightConst);
         }
-        return scalarComparator.compare(leftConstant, rightConstant);
+        return scalarComparator.compare(leftConst, rightConst);
     }
 }
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/LogicalScalarBinaryComparator.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/LogicalScalarBinaryComparator.java
index f2f637a..c2ff8d7 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/LogicalScalarBinaryComparator.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/LogicalScalarBinaryComparator.java
@@ -19,9 +19,9 @@
 package org.apache.asterix.dataflow.data.nontagged.comparators;
 
 import static org.apache.asterix.dataflow.data.common.ILogicalBinaryComparator.inequalityUndefined;
-import static org.apache.asterix.om.types.ATypeTag.VALUE_TYPE_MAPPING;
 
 import org.apache.asterix.dataflow.data.common.ILogicalBinaryComparator;
+import org.apache.asterix.dataflow.data.common.TaggedValueReference;
 import org.apache.asterix.dataflow.data.nontagged.serde.ADateSerializerDeserializer;
 import org.apache.asterix.dataflow.data.nontagged.serde.ADateTimeSerializerDeserializer;
 import org.apache.asterix.dataflow.data.nontagged.serde.ADayTimeDurationSerializerDeserializer;
@@ -31,7 +31,6 @@
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.asterix.om.types.hierachy.ATypeHierarchy;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.data.std.api.IPointable;
 import org.apache.hyracks.data.std.primitive.BooleanPointable;
 import org.apache.hyracks.data.std.primitive.ByteArrayPointable;
 import org.apache.hyracks.data.std.primitive.UTF8StringPointable;
@@ -51,9 +50,9 @@
     }
 
     @Override
-    public Result compare(IPointable left, IPointable right) throws HyracksDataException {
-        ATypeTag leftTag = VALUE_TYPE_MAPPING[left.getByteArray()[left.getStartOffset()]];
-        ATypeTag rightTag = VALUE_TYPE_MAPPING[right.getByteArray()[right.getStartOffset()]];
+    public Result compare(TaggedValueReference left, TaggedValueReference right) throws HyracksDataException {
+        ATypeTag leftTag = left.getTag();
+        ATypeTag rightTag = right.getTag();
         Result comparisonResult = ComparatorUtil.returnMissingOrNullOrMismatch(leftTag, rightTag);
         if (comparisonResult != null) {
             return comparisonResult;
@@ -61,9 +60,14 @@
         if (comparisonUndefined(leftTag, rightTag, isEquality)) {
             return Result.INCOMPARABLE;
         }
+        byte[] leftBytes = left.getByteArray();
+        byte[] rightBytes = right.getByteArray();
+        int leftStart = left.getStartOffset();
+        int rightStart = right.getStartOffset();
         // compare number if one of args is number since compatibility has already been checked above
         if (ATypeHierarchy.getTypeDomain(leftTag) == ATypeHierarchy.Domain.NUMERIC) {
-            return ComparatorUtil.compareNumbers(leftTag, left, rightTag, right);
+            return ILogicalBinaryComparator.asResult(
+                    ComparatorUtil.compareNumbers(leftTag, leftBytes, leftStart, rightTag, rightBytes, rightStart));
         }
 
         // comparing non-numeric
@@ -72,13 +76,8 @@
             throw new IllegalStateException("Two different non-numeric tags but they are compatible");
         }
 
-        byte[] leftBytes = left.getByteArray();
-        byte[] rightBytes = right.getByteArray();
-        int leftStart = left.getStartOffset() + 1;
-        int rightStart = right.getStartOffset() + 1;
-        int leftLen = left.getLength() - 1;
-        int rightLen = right.getLength() - 1;
-
+        int leftLen = left.getLength();
+        int rightLen = right.getLength();
         int result;
         switch (leftTag) {
             case BOOLEAN:
@@ -153,9 +152,9 @@
     }
 
     @Override
-    public Result compare(IPointable left, IAObject rightConstant) {
+    public Result compare(TaggedValueReference left, IAObject rightConstant) {
         // TODO(ali): currently defined for numbers only
-        ATypeTag leftTag = VALUE_TYPE_MAPPING[left.getByteArray()[left.getStartOffset()]];
+        ATypeTag leftTag = left.getTag();
         ATypeTag rightTag = rightConstant.getType().getTypeTag();
         Result comparisonResult = ComparatorUtil.returnMissingOrNullOrMismatch(leftTag, rightTag);
         if (comparisonResult != null) {
@@ -165,13 +164,13 @@
             return Result.NULL;
         }
         if (ATypeHierarchy.getTypeDomain(leftTag) == ATypeHierarchy.Domain.NUMERIC) {
-            return ComparatorUtil.compareNumWithConstant(leftTag, left, rightConstant);
+            return ComparatorUtil.compareNumWithConstant(left, rightConstant);
         }
         return Result.NULL;
     }
 
     @Override
-    public Result compare(IAObject leftConstant, IPointable right) {
+    public Result compare(IAObject leftConstant, TaggedValueReference right) {
         // TODO(ali): currently defined for numbers only
         Result result = compare(right, leftConstant);
         if (result == Result.LT) {
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/hash/AMurmurHash3BinaryHashFunctionFamily.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/hash/AMurmurHash3BinaryHashFunctionFamily.java
index f02c764..caab273 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/hash/AMurmurHash3BinaryHashFunctionFamily.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/hash/AMurmurHash3BinaryHashFunctionFamily.java
@@ -26,6 +26,7 @@
 import java.io.IOException;
 
 import org.apache.asterix.dataflow.data.common.ListAccessorUtil;
+import org.apache.asterix.dataflow.data.nontagged.serde.AOrderedListSerializerDeserializer;
 import org.apache.asterix.om.pointables.nonvisitor.RecordField;
 import org.apache.asterix.om.pointables.nonvisitor.SortedRecord;
 import org.apache.asterix.om.typecomputer.impl.TypeComputeUtils;
@@ -138,7 +139,7 @@
             IAType arrayType = TypeComputeUtils.getActualTypeOrOpen(type, ATypeTag.ARRAY);
             IAType itemType = ((AbstractCollectionType) arrayType).getItemType();
             ATypeTag itemTag = itemType.getTypeTag();
-            int numItems = ListAccessorUtil.numberOfItems(bytes, offset);
+            int numItems = AOrderedListSerializerDeserializer.getNumberOfItems(bytes, offset);
             int hash = seed;
             IPointable item = voidPointableAllocator.allocate(null);
             ArrayBackedValueStorage storage = (ArrayBackedValueStorage) storageAllocator.allocate(null);
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AOrderedListSerializerDeserializer.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AOrderedListSerializerDeserializer.java
index 9307627..bffc3b2 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AOrderedListSerializerDeserializer.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AOrderedListSerializerDeserializer.java
@@ -116,32 +116,16 @@
         listBuilder.write(out, false);
     }
 
-    public static final int getOrderedListLength(byte[] serOrderedList, int offset) {
+    public static int getOrderedListLength(byte[] serOrderedList, int offset) {
         return AInt32SerializerDeserializer.getInt(serOrderedList, offset + 1);
     }
 
     public static int getNumberOfItems(byte[] serOrderedList, int offset) {
-        if (serOrderedList[offset] == ATypeTag.ARRAY.serialize()) {
-            // 6 = tag (1) + itemTag (1) + list size (4)
-            return AInt32SerializerDeserializer.getInt(serOrderedList, offset + 6);
-        } else {
-            return -1;
-        }
+        // 6 = tag (1) + itemTag (1) + list size (4)
+        return AInt32SerializerDeserializer.getInt(serOrderedList, offset + 6);
     }
 
     public static int getItemOffset(byte[] serOrderedList, int offset, int itemIndex) throws HyracksDataException {
-        if (serOrderedList[offset] == ATypeTag.ARRAY.serialize()) {
-            ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(serOrderedList[offset + 1]);
-            if (NonTaggedFormatUtil.isFixedSizedCollection(typeTag)) {
-                int length = NonTaggedFormatUtil.getFieldValueLength(serOrderedList, offset + 1, typeTag, true);
-                return offset + 10 + (length * itemIndex);
-            } else {
-                return offset + AInt32SerializerDeserializer.getInt(serOrderedList, offset + 10 + (4 * itemIndex));
-            }
-            // 10 = tag (1) + itemTag (1) + list size (4) + number of items (4)
-        } else {
-            return -1;
-        }
+        return SerializerDeserializerUtil.getItemOffset(serOrderedList, offset, itemIndex);
     }
-
 }
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AUnorderedListSerializerDeserializer.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AUnorderedListSerializerDeserializer.java
index 751146a..b5165d2 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AUnorderedListSerializerDeserializer.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AUnorderedListSerializerDeserializer.java
@@ -118,31 +118,16 @@
         listBuilder.write(out, false);
     }
 
-    public static final int getUnorderedListLength(byte[] serOrderedList, int offset) {
+    public static int getUnorderedListLength(byte[] serOrderedList, int offset) {
         return AInt32SerializerDeserializer.getInt(serOrderedList, offset + 1);
     }
 
     public static int getNumberOfItems(byte[] serOrderedList, int offset) {
-        if (serOrderedList[offset] == ATypeTag.MULTISET.serialize()) {
-            // 6 = tag (1) + itemTag (1) + list size (4)
-            return AInt32SerializerDeserializer.getInt(serOrderedList, offset + 6);
-        } else {
-            return -1;
-        }
+        // 6 = tag (1) + itemTag (1) + list size (4)
+        return AInt32SerializerDeserializer.getInt(serOrderedList, offset + 6);
     }
 
     public static int getItemOffset(byte[] serOrderedList, int offset, int itemIndex) throws HyracksDataException {
-        if (serOrderedList[offset] == ATypeTag.MULTISET.serialize()) {
-            ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(serOrderedList[offset + 1]);
-            if (NonTaggedFormatUtil.isFixedSizedCollection(typeTag)) {
-                int length = NonTaggedFormatUtil.getFieldValueLength(serOrderedList, offset + 1, typeTag, true);
-                return offset + 10 + (length * itemIndex);
-            } else {
-                return offset + AInt32SerializerDeserializer.getInt(serOrderedList, offset + 10 + (4 * itemIndex));
-            }
-            // 10 = tag (1) + itemTag (1) + list size (4) + number of items (4)
-        } else {
-            return -1;
-        }
+        return SerializerDeserializerUtil.getItemOffset(serOrderedList, offset, itemIndex);
     }
 }
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/SerializerDeserializerUtil.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/SerializerDeserializerUtil.java
index 4ea2c13..952ffb4 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/SerializerDeserializerUtil.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/SerializerDeserializerUtil.java
@@ -24,11 +24,13 @@
 
 import org.apache.asterix.common.exceptions.CompilationException;
 import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.dataflow.data.common.TaggedValueReference;
 import org.apache.asterix.om.base.AInt32;
 import org.apache.asterix.om.base.IAObject;
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.asterix.om.types.EnumDeserializer;
 import org.apache.asterix.om.types.IAType;
+import org.apache.asterix.om.utils.NonTaggedFormatUtil;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
@@ -98,4 +100,33 @@
         return castBuffer.getByteArray();
     }
 
+    public static int getNumberOfItemsNonTagged(TaggedValueReference list) {
+        // 5 = itemTag (1) + list size (4)
+        return AInt32SerializerDeserializer.getInt(list.getByteArray(), list.getStartOffset() + 5);
+    }
+
+    public static int getItemOffset(byte[] listBytes, int offset, int itemIndex) throws HyracksDataException {
+        ATypeTag itemTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(listBytes[offset + 1]);
+        // 10 = tag (1) + itemTag (1) + list size (4) + number of items (4)
+        if (NonTaggedFormatUtil.isFixedSizedCollection(itemTag)) {
+            int valueLength = NonTaggedFormatUtil.getFieldValueLength(listBytes, offset + 1, itemTag, true);
+            return offset + 10 + (valueLength * itemIndex);
+        } else {
+            return offset + AInt32SerializerDeserializer.getInt(listBytes, offset + 10 + (4 * itemIndex));
+        }
+    }
+
+    public static int getItemOffsetNonTagged(TaggedValueReference list, int itemIndex) throws HyracksDataException {
+        byte[] listValueBytes = list.getByteArray();
+        int offset = list.getStartOffset();
+        ATypeTag itemTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(listValueBytes[offset]);
+        // 9 = itemTag (1) + list size (4) + number of items (4)
+        if (NonTaggedFormatUtil.isFixedSizedCollection(itemTag)) {
+            int valueLength = NonTaggedFormatUtil.getFieldValueLength(listValueBytes, offset, itemTag, true);
+            return offset + 9 + (valueLength * itemIndex);
+        } else {
+            // the -1 is due to the fact that the item encoded offset is measured from a tagged list
+            return offset + AInt32SerializerDeserializer.getInt(listValueBytes, offset + 9 + (4 * itemIndex)) - 1;
+        }
+    }
 }
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/ARecordPointable.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/ARecordPointable.java
index 928c71a..af8d5e6 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/ARecordPointable.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/ARecordPointable.java
@@ -53,13 +53,13 @@
  *   int length;
  *   byte isExpanded?;
  *   int openOffset?;
- *   int numberOfClosedFields;
- *   byte[ceil (numberOfFields / 4)] nullBitMap; // 2 bits per field, "1" means field is null, "2" field is missing
- *   int[numberOfClosedFields] closedFieldOffset;
- *   IPointable[numberOfClosedFields] fieldValue;
+ *   int numberOfClosedFields?;
+ *   byte[ceil (numberOfFields / 4)] nullBitMap?; // 2 bits per field, "1" means field is null, "2" field is missing
+ *   int[numberOfClosedFields] closedFieldOffset?;
+ *   IPointable[numberOfClosedFields] fieldValues?;
  *   int numberOfOpenFields?;
- *   OpenFieldLookup[numberOfOpenFields] lookup;
- *   OpenField[numberOfOpenFields] openFields;
+ *   OpenFieldLookup[numberOfOpenFields] lookup?;
+ *   OpenField[numberOfOpenFields] openFields?;
  * }
  *
  * OpenFieldLookup {
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/RecordField.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/RecordField.java
index 0cd16dc..3c33f49 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/RecordField.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/RecordField.java
@@ -59,7 +59,7 @@
         return namePointable;
     }
 
-    final int getIndex() {
+    public final int getIndex() {
         return index;
     }
 
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/SortedRecord.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/SortedRecord.java
index 1d9ac6c..be81a70 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/SortedRecord.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/SortedRecord.java
@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.util.PriorityQueue;
 
+import org.apache.asterix.dataflow.data.common.TaggedValueReference;
 import org.apache.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
 import org.apache.asterix.om.typecomputer.impl.TypeComputeUtils;
 import org.apache.asterix.om.types.ARecordType;
@@ -97,48 +98,58 @@
      * closed and open. It populates the utf8 filed names for open.
      */
     public final void reset(byte[] data, int start) throws HyracksDataException {
+        resetRecord(data, start, ARecordPointable.TAG_SIZE, 0);
+    }
+
+    public final void resetNonTagged(byte[] data, int start) throws HyracksDataException {
+        resetRecord(data, start, 0, 1);
+    }
+
+    private void resetRecord(byte[] data, int start, int skipTag, int fixOffset) throws HyracksDataException {
         bytes = data;
         reset();
         boolean isExpanded = false;
         // advance to expanded byte if present
-        int cursor = start + ARecordPointable.TAG_SIZE + ARecordPointable.RECORD_LENGTH_SIZE;
+        int pointer = start + skipTag + ARecordPointable.RECORD_LENGTH_SIZE;
         if (recordType.isOpen()) {
-            isExpanded = bytes[cursor] == 1;
+            isExpanded = bytes[pointer] == 1;
             // advance to either open part offset or number of closed fields
-            cursor += ARecordPointable.EXPANDED_SIZE;
+            pointer += ARecordPointable.EXPANDED_SIZE;
         }
         int openPartOffset = 0;
         if (isExpanded) {
-            openPartOffset = start + AInt32SerializerDeserializer.getInt(bytes, cursor);
+            openPartOffset = start + AInt32SerializerDeserializer.getInt(bytes, pointer) - fixOffset;
             // advance to number of closed fields
-            cursor += ARecordPointable.OPEN_OFFSET_SIZE;
+            pointer += ARecordPointable.OPEN_OFFSET_SIZE;
         }
         int fieldOffset;
         int length;
         int fieldIndex = 0;
         ATypeTag tag;
         // advance to where fields offsets are (or to null bit map if the schema has optional fields)
-        cursor += ARecordPointable.CLOSED_COUNT_SIZE;
-        int nullBitMapOffset = cursor;
-        int fieldsOffsets = cursor + nullBitMapSize;
-        // compute the offsets of each closed field value and whether it's missing or null
-        for (int i = 0; i < numSchemaFields; i++, fieldIndex++) {
-            fieldOffset = AInt32SerializerDeserializer.getInt(bytes, fieldsOffsets) + start;
-            tag = TypeComputeUtils.getActualType(fieldTypes[i]).getTypeTag();
-            if (hasOptionalFields) {
-                byte nullBits = bytes[nullBitMapOffset + i / 4];
-                if (RecordUtil.isNull(nullBits, i)) {
-                    tag = ATypeTag.NULL;
-                } else if (RecordUtil.isMissing(nullBits, i)) {
-                    tag = ATypeTag.MISSING;
+        if (numSchemaFields > 0) {
+            pointer += ARecordPointable.CLOSED_COUNT_SIZE;
+            int nullBitMapOffset = pointer;
+            int fieldsOffsets = nullBitMapOffset + nullBitMapSize;
+            // compute the offsets of each closed field value and whether it's missing or null
+            for (int i = 0; i < numSchemaFields; i++, fieldIndex++) {
+                fieldOffset = AInt32SerializerDeserializer.getInt(bytes, fieldsOffsets) + start - fixOffset;
+                tag = TypeComputeUtils.getActualType(fieldTypes[i]).getTypeTag();
+                if (hasOptionalFields) {
+                    byte nullBits = bytes[nullBitMapOffset + i / 4];
+                    if (RecordUtil.isNull(nullBits, i)) {
+                        tag = ATypeTag.NULL;
+                    } else if (RecordUtil.isMissing(nullBits, i)) {
+                        tag = ATypeTag.MISSING;
+                    }
                 }
+                length = NonTaggedFormatUtil.getFieldValueLength(bytes, fieldOffset, tag, false);
+                closedFields[i].set(fieldIndex, fieldOffset, length, tag);
+                if (tag != ATypeTag.MISSING) {
+                    sortedFields.add(closedFields[i]);
+                }
+                fieldsOffsets += ARecordPointable.FIELD_OFFSET_SIZE;
             }
-            length = NonTaggedFormatUtil.getFieldValueLength(bytes, fieldOffset, tag, false);
-            closedFields[i].set(fieldIndex, fieldOffset, length, tag);
-            if (tag != ATypeTag.MISSING) {
-                sortedFields.add(closedFields[i]);
-            }
-            fieldsOffsets += ARecordPointable.FIELD_OFFSET_SIZE;
         }
         // then populate open fields info second, an open field has name + value (tagged)
         if (isExpanded) {
@@ -185,6 +196,21 @@
         return RecordUtil.getType(recordType, field.getIndex(), field.getValueTag());
     }
 
+    public final void getFieldValue(RecordField field, TaggedValueReference fieldValueRef) {
+        if (field.getIndex() >= numSchemaFields) {
+            fieldValueRef.set(bytes, field.getValueOffset() + 1, field.getValueLength() - 1, field.getValueTag());
+        } else {
+            if (field.getValueTag() == ATypeTag.MISSING) {
+                fieldValueRef.set(MISSING_BYTES, 0, 0, ATypeTag.MISSING);
+            } else if (field.getValueTag() == ATypeTag.NULL) {
+                fieldValueRef.set(NULL_BYTES, 0, 0, ATypeTag.NULL);
+            } else {
+                fieldValueRef.set(bytes, field.getValueOffset(), field.getValueLength(), field.getValueTag());
+            }
+        }
+    }
+
+    // TODO(ali): remove this method once hashing does not need the tag to be adjacent to the value
     public final void getFieldValue(RecordField field, IPointable pointable, ArrayBackedValueStorage storage)
             throws IOException {
         int fieldIdx = field.getIndex();
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/util/container/ObjectFactories.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/util/container/ObjectFactories.java
index 2c6e408..ff76ea3 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/util/container/ObjectFactories.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/util/container/ObjectFactories.java
@@ -20,6 +20,7 @@
 
 import java.util.BitSet;
 
+import org.apache.asterix.dataflow.data.common.TaggedValueReference;
 import org.apache.asterix.om.pointables.nonvisitor.SortedRecord;
 import org.apache.asterix.om.types.ARecordType;
 import org.apache.hyracks.data.std.api.IMutableValueStorage;
@@ -45,4 +46,5 @@
     public static final IObjectFactory<BitSet, Void> BIT_SET_FACTORY = type -> new BitSet();
     public static final IObjectFactory<UTF8StringPointable, Void> UTF8_FACTORY = type -> new UTF8StringPointable();
     public static final IObjectFactory<SortedRecord, ARecordType> RECORD_FACTORY = SortedRecord::new;
+    public static final IObjectFactory<TaggedValueReference, Void> VALUE_FACTORY = type -> new TaggedValueReference();
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractMinMaxAggregateFunction.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractMinMaxAggregateFunction.java
index aa45922..2c49f71 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractMinMaxAggregateFunction.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractMinMaxAggregateFunction.java
@@ -18,11 +18,14 @@
  */
 package org.apache.asterix.runtime.aggregates.std;
 
+import static org.apache.asterix.om.types.ATypeTag.VALUE_TYPE_MAPPING;
+
 import java.io.IOException;
 
 import org.apache.asterix.common.exceptions.ErrorCode;
 import org.apache.asterix.common.exceptions.WarningUtil;
 import org.apache.asterix.dataflow.data.common.ILogicalBinaryComparator;
+import org.apache.asterix.dataflow.data.common.TaggedValueReference;
 import org.apache.asterix.dataflow.data.nontagged.comparators.ComparatorUtil;
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.asterix.om.types.EnumDeserializer;
@@ -45,6 +48,8 @@
     private final IPointable inputVal = new VoidPointable();
     private final ArrayBackedValueStorage outputVal = new ArrayBackedValueStorage();
     private final ArrayBackedValueStorage tempValForCasting = new ArrayBackedValueStorage();
+    private final TaggedValueReference value1 = new TaggedValueReference();
+    private final TaggedValueReference value2 = new TaggedValueReference();
     private final IScalarEvaluator eval;
     private final boolean isMin;
     private final IAType aggFieldType;
@@ -170,7 +175,15 @@
     private void compareAndUpdate(ILogicalBinaryComparator c, IPointable newVal, ArrayBackedValueStorage currentVal,
             ATypeTag typeTag) throws HyracksDataException {
         // newVal is never NULL/MISSING here. it's already checked up. current value is the first encountered non-null.
-        ILogicalBinaryComparator.Result result = c.compare(newVal, currentVal);
+        byte[] newValByteArray = newVal.getByteArray();
+        int newValStartOffset = newVal.getStartOffset();
+        byte[] currentValByteArray = currentVal.getByteArray();
+        int currentValStartOffset = currentVal.getStartOffset();
+        value1.set(newValByteArray, newValStartOffset + 1, newVal.getLength() - 1,
+                VALUE_TYPE_MAPPING[newValByteArray[newValStartOffset]]);
+        value2.set(currentValByteArray, currentValStartOffset + 1, currentVal.getLength() - 1,
+                VALUE_TYPE_MAPPING[newValByteArray[newValStartOffset]]);
+        ILogicalBinaryComparator.Result result = c.compare(value1, value2);
         switch (result) {
             case LT:
                 if (isMin) {
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/comparisons/AbstractComparisonEvaluator.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/comparisons/AbstractComparisonEvaluator.java
index 216259d..244ffc3 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/comparisons/AbstractComparisonEvaluator.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/comparisons/AbstractComparisonEvaluator.java
@@ -18,10 +18,13 @@
  */
 package org.apache.asterix.runtime.evaluators.comparisons;
 
+import static org.apache.asterix.om.types.ATypeTag.VALUE_TYPE_MAPPING;
+
 import java.io.DataOutput;
 
 import org.apache.asterix.dataflow.data.common.ILogicalBinaryComparator;
 import org.apache.asterix.dataflow.data.common.ILogicalBinaryComparator.Result;
+import org.apache.asterix.dataflow.data.common.TaggedValueReference;
 import org.apache.asterix.dataflow.data.nontagged.comparators.ComparatorUtil;
 import org.apache.asterix.dataflow.data.nontagged.serde.ADoubleSerializerDeserializer;
 import org.apache.asterix.dataflow.data.nontagged.serde.AFloatSerializerDeserializer;
@@ -68,6 +71,8 @@
     protected final DataOutput out = resultStorage.getDataOutput();
     protected final TaggedValuePointable argLeft = TaggedValuePointable.FACTORY.createPointable();
     private final TaggedValuePointable argRight = TaggedValuePointable.FACTORY.createPointable();
+    private final TaggedValueReference leftVal = new TaggedValueReference();
+    private final TaggedValueReference rightVal = new TaggedValueReference();
     private final IScalarEvaluator evalLeft;
     private final IScalarEvaluator evalRight;
     protected final SourceLocation sourceLoc;
@@ -102,38 +107,41 @@
         if (PointableHelper.checkAndSetMissingOrNull(result, argLeft, argRight)) {
             return;
         }
-
+        leftVal.set(argLeft.getByteArray(), argLeft.getStartOffset() + 1, argLeft.getLength() - 1,
+                VALUE_TYPE_MAPPING[argLeft.getTag()]);
+        rightVal.set(argRight.getByteArray(), argRight.getStartOffset() + 1, argRight.getLength() - 1,
+                VALUE_TYPE_MAPPING[argRight.getTag()]);
         evaluateImpl(result);
     }
 
     protected abstract void evaluateImpl(IPointable result) throws HyracksDataException;
 
-    Result compare() throws HyracksDataException {
+    final Result compare() throws HyracksDataException {
         if (leftConstant != null) {
             if (rightConstant != null) {
                 // both are constants
                 return logicalComparator.compare(leftConstant, rightConstant);
             } else {
                 // left is constant, right isn't
-                return logicalComparator.compare(leftConstant, argRight);
+                return logicalComparator.compare(leftConstant, rightVal);
             }
         } else {
             if (rightConstant != null) {
                 // right is constant, left isn't
-                return logicalComparator.compare(argLeft, rightConstant);
+                return logicalComparator.compare(leftVal, rightConstant);
             } else {
-                return logicalComparator.compare(argLeft, argRight);
+                return logicalComparator.compare(leftVal, rightVal);
             }
         }
     }
 
-    void writeMissing(IPointable result) throws HyracksDataException {
+    final void writeMissing(IPointable result) throws HyracksDataException {
         resultStorage.reset();
         missingSerde.serialize(AMissing.MISSING, out);
         result.set(resultStorage);
     }
 
-    void writeNull(IPointable result) throws HyracksDataException {
+    final void writeNull(IPointable result) throws HyracksDataException {
         resultStorage.reset();
         nullSerde.serialize(ANull.NULL, out);
         result.set(resultStorage);