diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/ABinaryComparator.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/ABinaryComparator.java
index ec337eb..05ffb3a 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/ABinaryComparator.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/comparators/ABinaryComparator.java
@@ -30,19 +30,19 @@
  */
 public abstract class ABinaryComparator implements IBinaryComparator {
 
-    public static ComparableResultCode isComparable(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
+    public static ComparableResultCode isComparable(byte tag1,byte tag2) {
         // NULL Check. If one type is NULL, then we return NULL
-        if (b1[s1] == ATypeTag.SERIALIZED_NULL_TYPE_TAG || b2[s2] == ATypeTag.SERIALIZED_NULL_TYPE_TAG || b1[s1] == 0
-                || b2[s2] == 0) {
+        if (tag1 == ATypeTag.SERIALIZED_NULL_TYPE_TAG || tag2 == ATypeTag.SERIALIZED_NULL_TYPE_TAG || tag1 == 0
+                || tag2 == 0) {
             return ComparableResultCode.UNKNOWN;
         }
 
         // Checks whether two types are comparable or not
-        ATypeTag tag1 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b1[s1]);
-        ATypeTag tag2 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(b2[s2]);
+        ATypeTag typeTag1 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(tag1);
+        ATypeTag typeTag2 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(tag2);
 
         // Are two types compatible, meaning that they can be compared? (e.g., compare between numeric types
-        if (ATypeHierarchy.isCompatible(tag1, tag2)) {
+        if (ATypeHierarchy.isCompatible(typeTag1, typeTag2)) {
             return ComparableResultCode.TRUE;
         } else {
             return ComparableResultCode.FALSE;
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/AIntervalPointable.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/AIntervalPointable.java
new file mode 100644
index 0000000..e9b1578
--- /dev/null
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/pointables/nonvisitor/AIntervalPointable.java
@@ -0,0 +1,162 @@
+/*
+ * 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.om.pointables.nonvisitor;
+
+import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.om.types.EnumDeserializer;
+import org.apache.asterix.om.util.container.IObjectFactory;
+import org.apache.hyracks.api.dataflow.value.ITypeTraits;
+import org.apache.hyracks.data.std.api.AbstractPointable;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.api.IPointableFactory;
+import org.apache.hyracks.data.std.primitive.BytePointable;
+import org.apache.hyracks.data.std.primitive.IntegerPointable;
+import org.apache.hyracks.data.std.primitive.LongPointable;
+
+/*
+ * This class interprets the binary data representation of an interval.
+ *
+ * Interval {
+ *   byte type;
+ *   IPointable start;
+ *   IPointable end;
+ * }
+ */
+public class AIntervalPointable extends AbstractPointable {
+
+    public static final ITypeTraits TYPE_TRAITS = new ITypeTraits() {
+        private static final long serialVersionUID = 1L;
+
+        @Override
+        public boolean isFixedLength() {
+            return false;
+        }
+
+        @Override
+        public int getFixedLength() {
+            return 0;
+        }
+    };
+
+    public static final IPointableFactory FACTORY = new IPointableFactory() {
+        private static final long serialVersionUID = 1L;
+
+        @Override
+        public IPointable createPointable() {
+            return new AIntervalPointable();
+        }
+
+        @Override
+        public ITypeTraits getTypeTraits() {
+            return TYPE_TRAITS;
+        }
+    };
+
+    public static final IObjectFactory<IPointable, ATypeTag> ALLOCATOR = new IObjectFactory<IPointable, ATypeTag>() {
+        @Override
+        public IPointable create(ATypeTag type) {
+            return new AIntervalPointable();
+        }
+    };
+
+    private static final int TAG_SIZE = 1;
+
+    public byte getType() {
+        return BytePointable.getByte(bytes, getTypeOffset());
+    }
+
+    public ATypeTag getTypeTag() {
+        return EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(getType());
+    }
+
+    public int getTypeOffset() {
+        return start;
+    }
+
+    public int getTypeSize() {
+        return TAG_SIZE;
+    }
+
+    public void getStart(IPointable start) throws AsterixException {
+        start.set(bytes, getStartOffset(), getStartSize());
+    }
+
+    public int getStartOffset() {
+        return getTypeOffset() + getTypeSize();
+    }
+
+    public int getStartSize() throws AsterixException {
+        switch (getTypeTag()) {
+            case DATE:
+            case TIME:
+                return Integer.BYTES;
+            case DATETIME:
+                return Long.BYTES;
+            default:
+                throw new AsterixException("Unsupported interval type: " + getTypeTag() + ".");
+        }
+    }
+
+    public long getStartValue() throws AsterixException {
+        switch (getTypeTag()) {
+            case DATE:
+            case TIME:
+                return IntegerPointable.getInteger(bytes, getStartOffset());
+            case DATETIME:
+                return LongPointable.getLong(bytes, getStartOffset());
+            default:
+                throw new AsterixException("Unsupported interval type: " + getTypeTag() + ".");
+        }
+    }
+
+    public void getEnd(IPointable start) throws AsterixException {
+        start.set(bytes, getEndOffset(), getEndSize());
+    }
+
+    public int getEndOffset() throws AsterixException {
+        return getStartOffset() + getStartSize();
+    }
+
+    public int getEndSize() throws AsterixException {
+        switch (getTypeTag()) {
+            case DATE:
+            case TIME:
+                return Integer.BYTES;
+            case DATETIME:
+                return Long.BYTES;
+            default:
+                throw new AsterixException("Unsupported interval type: " + getTypeTag() + ".");
+        }
+    }
+
+    public long getEndValue() throws AsterixException {
+        switch (getTypeTag()) {
+            case DATE:
+            case TIME:
+                return IntegerPointable.getInteger(bytes, getEndOffset());
+            case DATETIME:
+                return LongPointable.getLong(bytes, getEndOffset());
+            default:
+                throw new AsterixException("Unsupported interval type: " + getTypeTag() + ".");
+        }
+    }
+
+}
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 c78cf6e..30f6562 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
@@ -21,22 +21,6 @@
 import java.io.DataOutput;
 
 import org.apache.asterix.dataflow.data.nontagged.comparators.ABinaryComparator;
-import org.apache.asterix.dataflow.data.nontagged.comparators.ACirclePartialBinaryComparatorFactory;
-import org.apache.asterix.dataflow.data.nontagged.comparators.ADurationPartialBinaryComparatorFactory;
-import org.apache.asterix.dataflow.data.nontagged.comparators.AIntervalAscPartialBinaryComparatorFactory;
-import org.apache.asterix.dataflow.data.nontagged.comparators.ALinePartialBinaryComparatorFactory;
-import org.apache.asterix.dataflow.data.nontagged.comparators.APoint3DPartialBinaryComparatorFactory;
-import org.apache.asterix.dataflow.data.nontagged.comparators.APointPartialBinaryComparatorFactory;
-import org.apache.asterix.dataflow.data.nontagged.comparators.APolygonPartialBinaryComparatorFactory;
-import org.apache.asterix.dataflow.data.nontagged.comparators.ARectanglePartialBinaryComparatorFactory;
-import org.apache.asterix.dataflow.data.nontagged.comparators.AUUIDPartialBinaryComparatorFactory;
-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;
-import org.apache.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
-import org.apache.asterix.dataflow.data.nontagged.serde.AInt64SerializerDeserializer;
-import org.apache.asterix.dataflow.data.nontagged.serde.AInt8SerializerDeserializer;
-import org.apache.asterix.formats.nontagged.AqlBinaryComparatorFactoryProvider;
 import org.apache.asterix.formats.nontagged.AqlSerializerDeserializerProvider;
 import org.apache.asterix.om.base.ABoolean;
 import org.apache.asterix.om.base.ANull;
@@ -48,14 +32,10 @@
 import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
 import org.apache.hyracks.api.context.IHyracksTaskContext;
 import org.apache.hyracks.api.dataflow.value.BinaryComparatorConstant.ComparableResultCode;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
 import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
 import org.apache.hyracks.data.std.api.IPointable;
-import org.apache.hyracks.data.std.primitive.ByteArrayPointable;
-import org.apache.hyracks.data.std.primitive.FloatPointable;
-import org.apache.hyracks.data.std.primitive.IntegerPointable;
+import org.apache.hyracks.data.std.primitive.TaggedValuePointable;
 import org.apache.hyracks.data.std.primitive.VoidPointable;
 import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
 import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
@@ -71,39 +51,20 @@
 
     protected ArrayBackedValueStorage resultStorage = new ArrayBackedValueStorage();
     protected DataOutput out = resultStorage.getDataOutput();
-    protected IPointable outLeft = new VoidPointable();
-    protected IPointable outRight = new VoidPointable();
+    protected TaggedValuePointable argLeft = (TaggedValuePointable) TaggedValuePointable.FACTORY.createPointable();
+    protected TaggedValuePointable argRight = (TaggedValuePointable) TaggedValuePointable.FACTORY.createPointable();
+    protected IPointable outLeft = VoidPointable.FACTORY.createPointable();
+    protected IPointable outRight = VoidPointable.FACTORY.createPointable();
     protected IScalarEvaluator evalLeft;
     protected IScalarEvaluator evalRight;
+    private ComparisonHelper ch = new ComparisonHelper();
+
     @SuppressWarnings("unchecked")
     protected ISerializerDeserializer<ABoolean> serde = AqlSerializerDeserializerProvider.INSTANCE
             .getSerializerDeserializer(BuiltinType.ABOOLEAN);
     @SuppressWarnings("unchecked")
     protected ISerializerDeserializer<ANull> nullSerde = AqlSerializerDeserializerProvider.INSTANCE
             .getSerializerDeserializer(BuiltinType.ANULL);
-    protected IBinaryComparator strBinaryComp = AqlBinaryComparatorFactoryProvider.UTF8STRING_POINTABLE_INSTANCE
-            .createBinaryComparator();
-    protected IBinaryComparator circleBinaryComp = ACirclePartialBinaryComparatorFactory.INSTANCE
-            .createBinaryComparator();
-    protected IBinaryComparator durationBinaryComp = ADurationPartialBinaryComparatorFactory.INSTANCE
-            .createBinaryComparator();
-    protected IBinaryComparator intervalBinaryComp = AIntervalAscPartialBinaryComparatorFactory.INSTANCE
-            .createBinaryComparator();
-    protected IBinaryComparator lineBinaryComparator = ALinePartialBinaryComparatorFactory.INSTANCE
-            .createBinaryComparator();
-    protected IBinaryComparator pointBinaryComparator = APointPartialBinaryComparatorFactory.INSTANCE
-            .createBinaryComparator();
-    protected IBinaryComparator point3DBinaryComparator = APoint3DPartialBinaryComparatorFactory.INSTANCE
-            .createBinaryComparator();
-    protected IBinaryComparator polygonBinaryComparator = APolygonPartialBinaryComparatorFactory.INSTANCE
-            .createBinaryComparator();
-    protected IBinaryComparator rectangleBinaryComparator = ARectanglePartialBinaryComparatorFactory.INSTANCE
-            .createBinaryComparator();
-    protected IBinaryComparator uuidBinaryComparator = AUUIDPartialBinaryComparatorFactory.INSTANCE
-            .createBinaryComparator();
-
-    protected final IBinaryComparator byteArrayComparator = new PointableBinaryComparatorFactory(
-            ByteArrayPointable.FACTORY).createBinaryComparator();
 
     public AbstractComparisonEvaluator(IScalarEvaluatorFactory evalLeftFactory,
             IScalarEvaluatorFactory evalRightFactory, IHyracksTaskContext context) throws AlgebricksException {
@@ -111,17 +72,71 @@
         this.evalRight = evalRightFactory.createScalarEvaluator(context);
     }
 
+    @Override
+    public void evaluate(IFrameTupleReference tuple, IPointable result) throws AlgebricksException {
+        resultStorage.reset();
+        evalInputs(tuple);
+
+        // checks whether we can apply >, >=, <, and <= to the given type since
+        // these operations cannot be defined for certain types.
+        if (isTotallyOrderable()) {
+            checkTotallyOrderable();
+        }
+
+        // Checks whether two types are comparable
+        switch (comparabilityCheck()) {
+            case UNKNOWN:
+                // result:UNKNOWN - NULL value found
+                try {
+                    nullSerde.serialize(ANull.NULL, out);
+                    result.set(resultStorage);
+                    return;
+                } catch (HyracksDataException e) {
+                    throw new AlgebricksException(e);
+                }
+            case FALSE:
+                // result:FALSE - two types cannot be compared. Thus we return FALSE since this is equality comparison
+                ABoolean b = ABoolean.FALSE;
+                try {
+                    serde.serialize(b, out);
+                } catch (HyracksDataException e) {
+                    throw new AlgebricksException(e);
+                }
+                break;
+            case TRUE:
+                // Two types can be compared
+                ComparisonResult r = compareResults();
+                ABoolean b1 = getComparisonResult(r) ? ABoolean.TRUE : ABoolean.FALSE;
+                try {
+                    serde.serialize(b1, out);
+                } catch (HyracksDataException e) {
+                    throw new AlgebricksException(e);
+                }
+                break;
+            default:
+                throw new AlgebricksException(
+                        "Comparison cannot be processed. The return code from ComparabilityCheck is not correct.");
+        }
+        result.set(resultStorage);
+    }
+
+    protected abstract boolean isTotallyOrderable();
+
+    protected abstract boolean getComparisonResult(ComparisonResult r);
+
     protected void evalInputs(IFrameTupleReference tuple) throws AlgebricksException {
-        evalLeft.evaluate(tuple, outLeft);
-        evalRight.evaluate(tuple, outRight);
+        evalLeft.evaluate(tuple, argLeft);
+        evalRight.evaluate(tuple, argRight);
+
+        argLeft.getValue(outLeft);
+        argRight.getValue(outRight);
     }
 
     // checks whether we can apply >, >=, <, and <= operations to the given type since
     // these operations can not be defined for certain types.
     protected void checkTotallyOrderable() throws AlgebricksException {
-        if (outLeft.getLength() != 0) {
-            ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER
-                    .deserialize(outLeft.getByteArray()[outLeft.getStartOffset()]);
+        if (argLeft.getLength() != 0) {
+            ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argLeft.getTag());
             switch (typeTag) {
                 case DURATION:
                 case INTERVAL:
@@ -142,8 +157,7 @@
     // checks whether two types are comparable
     protected ComparableResultCode comparabilityCheck() {
         // just check TypeTags
-        return ABinaryComparator.isComparable(outLeft.getByteArray(), outLeft.getStartOffset(), 1,
-                outRight.getByteArray(), outRight.getStartOffset(), 1);
+        return ABinaryComparator.isComparable(argLeft.getTag(), argRight.getTag());
     }
 
     protected ComparisonResult compareResults() throws AlgebricksException {
@@ -152,15 +166,10 @@
         ATypeTag typeTag1 = null;
         ATypeTag typeTag2 = null;
 
-        byte[] leftBytes = outLeft.getByteArray();
-        int leftStartOffset = outLeft.getStartOffset();
-        byte[] rightBytes = outRight.getByteArray();
-        int rightStartOffset = outRight.getStartOffset();
-
         if (outLeft.getLength() == 0) {
             isLeftNull = true;
         } else {
-            typeTag1 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(leftBytes[leftStartOffset]);
+            typeTag1 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argLeft.getTag());
             if (typeTag1 == ATypeTag.NULL) {
                 isLeftNull = true;
             }
@@ -168,7 +177,7 @@
         if (outRight.getLength() == 0) {
             isRightNull = true;
         } else {
-            typeTag2 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(rightBytes[rightStartOffset]);
+            typeTag2 = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argRight.getTag());
             if (typeTag2 == ATypeTag.NULL) {
                 isRightNull = true;
             }
@@ -178,111 +187,7 @@
             return ComparisonResult.UNKNOWN;
         }
 
-        switch (typeTag1) {
-            case INT8: {
-                return compareInt8WithArg(typeTag2);
-            }
-            case INT16: {
-                return compareInt16WithArg(typeTag2);
-            }
-            case INT32: {
-                return compareInt32WithArg(typeTag2);
-            }
-            case INT64: {
-                return compareInt64WithArg(typeTag2);
-            }
-            case FLOAT: {
-                return compareFloatWithArg(typeTag2);
-            }
-            case DOUBLE: {
-                return compareDoubleWithArg(typeTag2);
-            }
-            case STRING: {
-                return compareStringWithArg(typeTag2);
-            }
-            case BOOLEAN: {
-                return compareBooleanWithArg(typeTag2);
-            }
-
-            default: {
-                return compareStrongTypedWithArg(typeTag1, typeTag2);
-            }
-        }
-    }
-
-    private ComparisonResult compareStrongTypedWithArg(ATypeTag expectedTypeTag, ATypeTag actualTypeTag)
-            throws AlgebricksException {
-        if (expectedTypeTag != actualTypeTag) {
-            throw new AlgebricksException(
-                    "Comparison is undefined between " + expectedTypeTag + " and " + actualTypeTag + ".");
-        }
-        int result = 0;
-        byte[] leftBytes = outLeft.getByteArray();
-        int leftOffset = outLeft.getStartOffset() + 1;
-        int leftLen = outLeft.getLength() - 1;
-        byte[] rightBytes = outRight.getByteArray();
-        int rightOffset = outRight.getStartOffset() + 1;
-        int rightLen = outRight.getLength() - 1;
-
-        try {
-            switch (actualTypeTag) {
-                case YEARMONTHDURATION:
-                case TIME:
-                case DATE:
-                    result = Integer.compare(AInt32SerializerDeserializer.getInt(leftBytes, leftOffset),
-                            AInt32SerializerDeserializer.getInt(rightBytes, rightOffset));
-                    break;
-                case DAYTIMEDURATION:
-                case DATETIME:
-                    result = Long.compare(AInt64SerializerDeserializer.getLong(leftBytes, leftOffset),
-                            AInt64SerializerDeserializer.getLong(rightBytes, rightOffset));
-                    break;
-                case CIRCLE:
-                    result = circleBinaryComp.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
-                            rightLen);
-                    break;
-                case LINE:
-                    result = lineBinaryComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
-                            rightLen);
-                    break;
-                case POINT:
-                    result = pointBinaryComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
-                            rightLen);
-                    break;
-                case POINT3D:
-                    result = point3DBinaryComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
-                            rightLen);
-                    break;
-                case POLYGON:
-                    result = polygonBinaryComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
-                            rightLen);
-                    break;
-                case DURATION:
-                    result = durationBinaryComp.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
-                            rightLen);
-                    break;
-                case INTERVAL:
-                    result = intervalBinaryComp.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
-                            rightLen);
-                    break;
-                case RECTANGLE:
-                    result = rectangleBinaryComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
-                            rightLen);
-                    break;
-                case BINARY:
-                    result = byteArrayComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
-                            rightLen);
-                    break;
-                case UUID:
-                    result = uuidBinaryComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
-                            rightLen);
-                    break;
-                default:
-                    throw new AlgebricksException("Comparison for " + actualTypeTag + " is not supported.");
-            }
-        } catch (HyracksDataException e) {
-            throw new AlgebricksException(e);
-        }
+        int result = ch.compare(typeTag1, typeTag2, outLeft, outRight);
         if (result == 0) {
             return ComparisonResult.EQUAL;
         } else if (result < 0) {
@@ -292,322 +197,4 @@
         }
     }
 
-    private ComparisonResult compareBooleanWithArg(ATypeTag typeTag2) throws AlgebricksException {
-        if (typeTag2 == ATypeTag.BOOLEAN) {
-            byte b0 = outLeft.getByteArray()[outLeft.getStartOffset() + 1];
-            byte b1 = outRight.getByteArray()[outRight.getStartOffset() + 1];
-            return compareByte(b0, b1);
-        }
-        throw new AlgebricksException("Comparison is undefined between types ABoolean and " + typeTag2 + " .");
-    }
-
-    private ComparisonResult compareStringWithArg(ATypeTag typeTag2) throws AlgebricksException {
-        if (typeTag2 == ATypeTag.STRING) {
-            int result;
-            try {
-                result = strBinaryComp.compare(outLeft.getByteArray(), outLeft.getStartOffset() + 1,
-                        outLeft.getLength() - 1, outRight.getByteArray(), outRight.getStartOffset() + 1,
-                        outRight.getLength() - 1);
-            } catch (HyracksDataException e) {
-                throw new AlgebricksException(e);
-            }
-            if (result == 0) {
-                return ComparisonResult.EQUAL;
-            } else if (result < 0) {
-                return ComparisonResult.LESS_THAN;
-            } else {
-                return ComparisonResult.GREATER_THAN;
-            }
-        }
-        throw new AlgebricksException("Comparison is undefined between types AString and " + typeTag2 + " .");
-    }
-
-    private ComparisonResult compareDoubleWithArg(ATypeTag typeTag2) throws AlgebricksException {
-        byte[] leftBytes = outLeft.getByteArray();
-        int leftOffset = outLeft.getStartOffset() + 1;
-        byte[] rightBytes = outRight.getByteArray();
-        int rightOffset = outRight.getStartOffset() + 1;
-
-        double s = ADoubleSerializerDeserializer.getDouble(leftBytes, leftOffset);
-        switch (typeTag2) {
-            case INT8: {
-                byte v2 = AInt8SerializerDeserializer.getByte(rightBytes, rightOffset);
-                return compareDouble(s, v2);
-            }
-            case INT16: {
-                short v2 = AInt16SerializerDeserializer.getShort(rightBytes, rightOffset);
-                return compareDouble(s, v2);
-            }
-            case INT32: {
-                int v2 = AInt32SerializerDeserializer.getInt(rightBytes, rightOffset);
-                return compareDouble(s, v2);
-            }
-            case INT64: {
-                long v2 = AInt64SerializerDeserializer.getLong(rightBytes, rightOffset);
-                return compareDouble(s, v2);
-            }
-            case FLOAT: {
-                float v2 = AFloatSerializerDeserializer.getFloat(rightBytes, rightOffset);
-                return compareDouble(s, v2);
-            }
-            case DOUBLE: {
-                double v2 = ADoubleSerializerDeserializer.getDouble(rightBytes, rightOffset);
-                return compareDouble(s, v2);
-            }
-            default: {
-                throw new AlgebricksException("Comparison is undefined between types ADouble and " + typeTag2 + " .");
-            }
-        }
-    }
-
-    private ComparisonResult compareFloatWithArg(ATypeTag typeTag2) throws AlgebricksException {
-        byte[] leftBytes = outLeft.getByteArray();
-        int leftOffset = outLeft.getStartOffset() + 1;
-        byte[] rightBytes = outRight.getByteArray();
-        int rightOffset = outRight.getStartOffset() + 1;
-
-        float s = FloatPointable.getFloat(leftBytes, leftOffset);
-        switch (typeTag2) {
-            case INT8: {
-                byte v2 = AInt8SerializerDeserializer.getByte(rightBytes, rightOffset);
-                return compareFloat(s, v2);
-            }
-            case INT16: {
-                short v2 = AInt16SerializerDeserializer.getShort(rightBytes, rightOffset);
-                return compareFloat(s, v2);
-            }
-            case INT32: {
-                int v2 = AInt32SerializerDeserializer.getInt(rightBytes, rightOffset);
-                return compareFloat(s, v2);
-            }
-            case INT64: {
-                long v2 = AInt64SerializerDeserializer.getLong(rightBytes, rightOffset);
-                return compareFloat(s, v2);
-            }
-            case FLOAT: {
-                float v2 = AFloatSerializerDeserializer.getFloat(rightBytes, rightOffset);
-                return compareFloat(s, v2);
-            }
-            case DOUBLE: {
-                double v2 = ADoubleSerializerDeserializer.getDouble(rightBytes, rightOffset);
-                return compareDouble(s, v2);
-            }
-            default: {
-                throw new AlgebricksException("Comparison is undefined between types AFloat and " + typeTag2 + " .");
-            }
-        }
-    }
-
-    private ComparisonResult compareInt64WithArg(ATypeTag typeTag2) throws AlgebricksException {
-        byte[] leftBytes = outLeft.getByteArray();
-        int leftOffset = outLeft.getStartOffset() + 1;
-        byte[] rightBytes = outRight.getByteArray();
-        int rightOffset = outRight.getStartOffset() + 1;
-
-        long s = AInt64SerializerDeserializer.getLong(leftBytes, leftOffset);
-        switch (typeTag2) {
-            case INT8: {
-                byte v2 = AInt8SerializerDeserializer.getByte(rightBytes, rightOffset);
-                return compareLong(s, v2);
-            }
-            case INT16: {
-                short v2 = AInt16SerializerDeserializer.getShort(rightBytes, rightOffset);
-                return compareLong(s, v2);
-            }
-            case INT32: {
-                int v2 = AInt32SerializerDeserializer.getInt(rightBytes, rightOffset);
-                return compareLong(s, v2);
-            }
-            case INT64: {
-                long v2 = AInt64SerializerDeserializer.getLong(rightBytes, rightOffset);
-                return compareLong(s, v2);
-            }
-            case FLOAT: {
-                float v2 = AFloatSerializerDeserializer.getFloat(rightBytes, rightOffset);
-                return compareFloat(s, v2);
-            }
-            case DOUBLE: {
-                double v2 = ADoubleSerializerDeserializer.getDouble(rightBytes, rightOffset);
-                return compareDouble(s, v2);
-            }
-            default: {
-                throw new AlgebricksException("Comparison is undefined between types AInt64 and " + typeTag2 + " .");
-            }
-        }
-    }
-
-    private ComparisonResult compareInt32WithArg(ATypeTag typeTag2) throws AlgebricksException {
-        byte[] leftBytes = outLeft.getByteArray();
-        int leftOffset = outLeft.getStartOffset() + 1;
-        byte[] rightBytes = outRight.getByteArray();
-        int rightOffset = outRight.getStartOffset() + 1;
-
-        int s = IntegerPointable.getInteger(leftBytes, leftOffset);
-        switch (typeTag2) {
-            case INT8: {
-                byte v2 = AInt8SerializerDeserializer.getByte(rightBytes, rightOffset);
-                return compareInt(s, v2);
-            }
-            case INT16: {
-                short v2 = AInt16SerializerDeserializer.getShort(rightBytes, rightOffset);
-                return compareInt(s, v2);
-            }
-            case INT32: {
-                int v2 = AInt32SerializerDeserializer.getInt(rightBytes, rightOffset);
-                return compareInt(s, v2);
-            }
-            case INT64: {
-                long v2 = AInt64SerializerDeserializer.getLong(rightBytes, rightOffset);
-                return compareLong(s, v2);
-            }
-            case FLOAT: {
-                float v2 = AFloatSerializerDeserializer.getFloat(rightBytes, rightOffset);
-                return compareFloat(s, v2);
-            }
-            case DOUBLE: {
-                double v2 = ADoubleSerializerDeserializer.getDouble(rightBytes, rightOffset);
-                return compareDouble(s, v2);
-            }
-            default: {
-                throw new AlgebricksException("Comparison is undefined between types AInt32 and " + typeTag2 + " .");
-            }
-        }
-    }
-
-    private ComparisonResult compareInt16WithArg(ATypeTag typeTag2) throws AlgebricksException {
-        byte[] leftBytes = outLeft.getByteArray();
-        int leftOffset = outLeft.getStartOffset() + 1;
-        byte[] rightBytes = outRight.getByteArray();
-        int rightOffset = outRight.getStartOffset() + 1;
-
-        short s = AInt16SerializerDeserializer.getShort(leftBytes, leftOffset);
-        switch (typeTag2) {
-            case INT8: {
-                byte v2 = AInt8SerializerDeserializer.getByte(rightBytes, rightOffset);
-                return compareShort(s, v2);
-            }
-            case INT16: {
-                short v2 = AInt16SerializerDeserializer.getShort(rightBytes, rightOffset);
-                return compareShort(s, v2);
-            }
-            case INT32: {
-                int v2 = AInt32SerializerDeserializer.getInt(rightBytes, rightOffset);
-                return compareInt(s, v2);
-            }
-            case INT64: {
-                long v2 = AInt64SerializerDeserializer.getLong(rightBytes, rightOffset);
-                return compareLong(s, v2);
-            }
-            case FLOAT: {
-                float v2 = AFloatSerializerDeserializer.getFloat(rightBytes, rightOffset);
-                return compareFloat(s, v2);
-            }
-            case DOUBLE: {
-                double v2 = ADoubleSerializerDeserializer.getDouble(rightBytes, rightOffset);
-                return compareDouble(s, v2);
-            }
-            default: {
-                throw new AlgebricksException("Comparison is undefined between types AInt16 and " + typeTag2 + " .");
-            }
-        }
-    }
-
-    private ComparisonResult compareInt8WithArg(ATypeTag typeTag2) throws AlgebricksException {
-        byte[] leftBytes = outLeft.getByteArray();
-        int leftStart = outLeft.getStartOffset() + 1;
-        byte[] rightBytes = outRight.getByteArray();
-        int rightStart = outRight.getStartOffset() + 1;
-
-        byte s = AInt8SerializerDeserializer.getByte(leftBytes, leftStart);
-        switch (typeTag2) {
-            case INT8: {
-                byte v2 = AInt8SerializerDeserializer.getByte(rightBytes, rightStart);
-                return compareByte(s, v2);
-            }
-            case INT16: {
-                short v2 = AInt16SerializerDeserializer.getShort(rightBytes, rightStart);
-                return compareShort(s, v2);
-            }
-            case INT32: {
-                int v2 = AInt32SerializerDeserializer.getInt(rightBytes, rightStart);
-                return compareInt(s, v2);
-            }
-            case INT64: {
-                long v2 = AInt64SerializerDeserializer.getLong(rightBytes, rightStart);
-                return compareLong(s, v2);
-            }
-            case FLOAT: {
-                float v2 = AFloatSerializerDeserializer.getFloat(rightBytes, rightStart);
-                return compareFloat(s, v2);
-            }
-            case DOUBLE: {
-                double v2 = ADoubleSerializerDeserializer.getDouble(rightBytes, rightStart);
-                return compareDouble(s, v2);
-            }
-            default: {
-                throw new AlgebricksException("Comparison is undefined between types AInt16 and " + typeTag2 + " .");
-            }
-        }
-    }
-
-    private final ComparisonResult compareByte(int v1, int v2) {
-        if (v1 == v2) {
-            return ComparisonResult.EQUAL;
-        } else if (v1 < v2) {
-            return ComparisonResult.LESS_THAN;
-        } else {
-            return ComparisonResult.GREATER_THAN;
-        }
-    }
-
-    private final ComparisonResult compareShort(int v1, int v2) {
-        if (v1 == v2) {
-            return ComparisonResult.EQUAL;
-        } else if (v1 < v2) {
-            return ComparisonResult.LESS_THAN;
-        } else {
-            return ComparisonResult.GREATER_THAN;
-        }
-    }
-
-    private final ComparisonResult compareInt(int v1, int v2) {
-        if (v1 == v2) {
-            return ComparisonResult.EQUAL;
-        } else if (v1 < v2) {
-            return ComparisonResult.LESS_THAN;
-        } else {
-            return ComparisonResult.GREATER_THAN;
-        }
-    }
-
-    private final ComparisonResult compareLong(long v1, long v2) {
-        if (v1 == v2) {
-            return ComparisonResult.EQUAL;
-        } else if (v1 < v2) {
-            return ComparisonResult.LESS_THAN;
-        } else {
-            return ComparisonResult.GREATER_THAN;
-        }
-    }
-
-    private final ComparisonResult compareFloat(float v1, float v2) {
-        if (v1 == v2) {
-            return ComparisonResult.EQUAL;
-        } else if (v1 < v2) {
-            return ComparisonResult.LESS_THAN;
-        } else {
-            return ComparisonResult.GREATER_THAN;
-        }
-    }
-
-    private final ComparisonResult compareDouble(double v1, double v2) {
-        if (v1 == v2) {
-            return ComparisonResult.EQUAL;
-        } else if (v1 < v2) {
-            return ComparisonResult.LESS_THAN;
-        } else {
-            return ComparisonResult.GREATER_THAN;
-        }
-    }
-
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/comparisons/ComparisonEvalFactory.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/comparisons/ComparisonEvalFactory.java
index e549345..649f262 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/comparisons/ComparisonEvalFactory.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/comparisons/ComparisonEvalFactory.java
@@ -18,16 +18,11 @@
  */
 package org.apache.asterix.runtime.evaluators.comparisons;
 
-import org.apache.asterix.om.base.ABoolean;
-import org.apache.asterix.om.base.ANull;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.AlgebricksBuiltinFunctions.ComparisonKind;
 import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
 import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
 import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.data.std.api.IPointable;
-import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
 
 public class ComparisonEvalFactory implements IScalarEvaluatorFactory {
     private static final long serialVersionUID = 1L;
@@ -78,46 +73,14 @@
         }
 
         @Override
-        public void evaluate(IFrameTupleReference tuple, IPointable result) throws AlgebricksException {
-            resultStorage.reset();
-            evalInputs(tuple);
-            // Checks whether two types are comparable
-            switch (comparabilityCheck()) {
-                case UNKNOWN:
-                    // result:UNKNOWN - NULL value found
-                    try {
-                        nullSerde.serialize(ANull.NULL, out);
-                        result.set(resultStorage);
-                        return;
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                case FALSE:
-                    // result:FALSE - two types cannot be compared. Thus we return FALSE since this is equality comparison
-                    ABoolean b = ABoolean.FALSE;
-                    try {
-                        serde.serialize(b, out);
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                    break;
-                case TRUE:
-                    // Two types can be compared
-                    ComparisonResult r = compareResults();
-                    ABoolean b1 = (r == ComparisonResult.EQUAL) ? ABoolean.TRUE : ABoolean.FALSE;
-                    try {
-                        serde.serialize(b1, out);
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                    break;
-                default:
-                    throw new AlgebricksException(
-                            "Equality Comparison cannot be processed. The return code from ComparabilityCheck is not correct.");
-            }
-            result.set(resultStorage);
+        protected boolean getComparisonResult(ComparisonResult r) {
+            return (r == ComparisonResult.EQUAL);
         }
 
+        @Override
+        protected boolean isTotallyOrderable() {
+            return false;
+        }
     }
 
     static class InequalityComparisonEvaluator extends AbstractComparisonEvaluator {
@@ -127,47 +90,14 @@
         }
 
         @Override
-        public void evaluate(IFrameTupleReference tuple, IPointable result) throws AlgebricksException {
-            resultStorage.reset();
-            evalInputs(tuple);
-
-            // Checks whether two types are comparable
-            switch (comparabilityCheck()) {
-                case UNKNOWN:
-                    // result:UNKNOWN - NULL value found
-                    try {
-                        nullSerde.serialize(ANull.NULL, out);
-                        result.set(resultStorage);
-                        return;
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                case FALSE:
-                    // result:FALSE - two types cannot be compared. Thus we return TRUE since this is NOT EQ comparison.
-                    ABoolean b = ABoolean.TRUE;
-                    try {
-                        serde.serialize(b, out);
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                    break;
-                case TRUE:
-                    // Two types can be compared
-                    ComparisonResult r = compareResults();
-                    ABoolean b1 = (r != ComparisonResult.EQUAL) ? ABoolean.TRUE : ABoolean.FALSE;
-                    try {
-                        serde.serialize(b1, out);
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                    break;
-                default:
-                    throw new AlgebricksException(
-                            "Inequality Comparison cannot be processed. The return code from ComparabilityCheck is not correct.");
-            }
-            result.set(resultStorage);
+        protected boolean getComparisonResult(ComparisonResult r) {
+            return (r != ComparisonResult.EQUAL);
         }
 
+        @Override
+        protected boolean isTotallyOrderable() {
+            return false;
+        }
     }
 
     static class GreaterThanOrEqualComparisonEvaluator extends AbstractComparisonEvaluator {
@@ -177,52 +107,14 @@
         }
 
         @Override
-        public void evaluate(IFrameTupleReference tuple, IPointable result) throws AlgebricksException {
-            resultStorage.reset();
-            evalInputs(tuple);
-
-            // checks whether we can apply >, >=, <, and <= to the given type since
-            // these operations cannot be defined for certain types.
-            checkTotallyOrderable();
-
-            // Checks whether two types are comparable
-            switch (comparabilityCheck()) {
-                case UNKNOWN:
-                    // result:UNKNOWN - NULL value found
-                    try {
-                        nullSerde.serialize(ANull.NULL, out);
-                        result.set(resultStorage);
-                        return;
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                case FALSE:
-                    // result:FALSE - two types cannot be compared. Thus we return FALSE since this is an inequality comparison.
-                    ABoolean b = ABoolean.FALSE;
-                    try {
-                        serde.serialize(b, out);
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                    break;
-                case TRUE:
-                    // Two types can be compared
-                    ComparisonResult r = compareResults();
-                    ABoolean b1 = (r == ComparisonResult.EQUAL || r == ComparisonResult.GREATER_THAN) ? ABoolean.TRUE
-                            : ABoolean.FALSE;
-                    try {
-                        serde.serialize(b1, out);
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                    break;
-                default:
-                    throw new AlgebricksException(
-                            "Inequality Comparison cannot be processed. The return code from ComparabilityCheck is not correct.");
-            }
-            result.set(resultStorage);
+        protected boolean getComparisonResult(ComparisonResult r) {
+            return (r == ComparisonResult.EQUAL || r == ComparisonResult.GREATER_THAN);
         }
 
+        @Override
+        protected boolean isTotallyOrderable() {
+            return true;
+        }
     }
 
     static class GreaterThanComparisonEvaluator extends AbstractComparisonEvaluator {
@@ -232,51 +124,14 @@
         }
 
         @Override
-        public void evaluate(IFrameTupleReference tuple, IPointable result) throws AlgebricksException {
-            resultStorage.reset();
-            evalInputs(tuple);
-
-            // checks whether we can apply >, >=, <, and <= to the given type since
-            // these operations cannot be defined for certain types.
-            checkTotallyOrderable();
-
-            // Checks whether two types are comparable
-            switch (comparabilityCheck()) {
-                case UNKNOWN:
-                    // result:UNKNOWN - NULL value found
-                    try {
-                        nullSerde.serialize(ANull.NULL, out);
-                        result.set(resultStorage);
-                        return;
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                case FALSE:
-                    // result:FALSE - two types cannot be compared. Thus we return FALSE since this is an inequality comparison.
-                    ABoolean b = ABoolean.FALSE;
-                    try {
-                        serde.serialize(b, out);
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                    break;
-                case TRUE:
-                    // Two types can be compared
-                    ComparisonResult r = compareResults();
-                    ABoolean b1 = (r == ComparisonResult.GREATER_THAN) ? ABoolean.TRUE : ABoolean.FALSE;
-                    try {
-                        serde.serialize(b1, out);
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                    break;
-                default:
-                    throw new AlgebricksException(
-                            "Inequality Comparison cannot be processed. The return code from ComparabilityCheck is not correct.");
-            }
-            result.set(resultStorage);
+        protected boolean getComparisonResult(ComparisonResult r) {
+            return (r == ComparisonResult.GREATER_THAN);
         }
 
+        @Override
+        protected boolean isTotallyOrderable() {
+            return true;
+        }
     }
 
     static class LessThanOrEqualComparisonEvaluator extends AbstractComparisonEvaluator {
@@ -286,52 +141,14 @@
         }
 
         @Override
-        public void evaluate(IFrameTupleReference tuple, IPointable result) throws AlgebricksException {
-            resultStorage.reset();
-            evalInputs(tuple);
-
-            // checks whether we can apply >, >=, <, and <= to the given type since
-            // these operations cannot be defined for certain types.
-            checkTotallyOrderable();
-
-            // Checks whether two types are comparable
-            switch (comparabilityCheck()) {
-                case UNKNOWN:
-                    // result:UNKNOWN - NULL value found
-                    try {
-                        nullSerde.serialize(ANull.NULL, out);
-                        result.set(resultStorage);
-                        return;
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                case FALSE:
-                    // result:FALSE - two types cannot be compared. Thus we return FALSE since this is an inequality comparison.
-                    ABoolean b = ABoolean.FALSE;
-                    try {
-                        serde.serialize(b, out);
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                    break;
-                case TRUE:
-                    // Two types can be compared
-                    ComparisonResult r = compareResults();
-                    ABoolean b1 = (r == ComparisonResult.EQUAL || r == ComparisonResult.LESS_THAN) ? ABoolean.TRUE
-                            : ABoolean.FALSE;
-                    try {
-                        serde.serialize(b1, out);
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                    break;
-                default:
-                    throw new AlgebricksException(
-                            "Inequality Comparison cannot be processed. The return code from ComparabilityCheck is not correct.");
-            }
-            result.set(resultStorage);
+        protected boolean getComparisonResult(ComparisonResult r) {
+            return (r == ComparisonResult.EQUAL || r == ComparisonResult.LESS_THAN);
         }
 
+        @Override
+        protected boolean isTotallyOrderable() {
+            return true;
+        }
     }
 
     static class LessThanComparisonEvaluator extends AbstractComparisonEvaluator {
@@ -341,51 +158,14 @@
         }
 
         @Override
-        public void evaluate(IFrameTupleReference tuple, IPointable result) throws AlgebricksException {
-            resultStorage.reset();
-            evalInputs(tuple);
-
-            // checks whether we can apply >, >=, <, and <= to the given type since
-            // these operations cannot be defined for certain types.
-            checkTotallyOrderable();
-
-            // Checks whether two types are comparable
-            switch (comparabilityCheck()) {
-                case UNKNOWN:
-                    // result:UNKNOWN - NULL value found
-                    try {
-                        nullSerde.serialize(ANull.NULL, out);
-                        result.set(resultStorage);
-                        return;
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                case FALSE:
-                    // result:FALSE - two types cannot be compared. Thus we return FALSE since this is an inequality comparison.
-                    ABoolean b = ABoolean.FALSE;
-                    try {
-                        serde.serialize(b, out);
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                    break;
-                case TRUE:
-                    // Two types can be compared
-                    ComparisonResult r = compareResults();
-                    ABoolean b1 = (r == ComparisonResult.LESS_THAN) ? ABoolean.TRUE : ABoolean.FALSE;
-                    try {
-                        serde.serialize(b1, out);
-                    } catch (HyracksDataException e) {
-                        throw new AlgebricksException(e);
-                    }
-                    break;
-                default:
-                    throw new AlgebricksException(
-                            "Inequality Comparison cannot be processed. The return code from ComparabilityCheck is not correct.");
-            }
-            result.set(resultStorage);
+        protected boolean getComparisonResult(ComparisonResult r) {
+            return (r == ComparisonResult.LESS_THAN);
         }
 
+        @Override
+        protected boolean isTotallyOrderable() {
+            return true;
+        }
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/comparisons/ComparisonHelper.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/comparisons/ComparisonHelper.java
new file mode 100644
index 0000000..f2be58f
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/comparisons/ComparisonHelper.java
@@ -0,0 +1,486 @@
+/*
+ * 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.runtime.evaluators.comparisons;
+
+import java.io.Serializable;
+
+import org.apache.asterix.dataflow.data.nontagged.comparators.ACirclePartialBinaryComparatorFactory;
+import org.apache.asterix.dataflow.data.nontagged.comparators.ADurationPartialBinaryComparatorFactory;
+import org.apache.asterix.dataflow.data.nontagged.comparators.AIntervalAscPartialBinaryComparatorFactory;
+import org.apache.asterix.dataflow.data.nontagged.comparators.ALinePartialBinaryComparatorFactory;
+import org.apache.asterix.dataflow.data.nontagged.comparators.APoint3DPartialBinaryComparatorFactory;
+import org.apache.asterix.dataflow.data.nontagged.comparators.APointPartialBinaryComparatorFactory;
+import org.apache.asterix.dataflow.data.nontagged.comparators.APolygonPartialBinaryComparatorFactory;
+import org.apache.asterix.dataflow.data.nontagged.comparators.ARectanglePartialBinaryComparatorFactory;
+import org.apache.asterix.dataflow.data.nontagged.comparators.AUUIDPartialBinaryComparatorFactory;
+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;
+import org.apache.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
+import org.apache.asterix.dataflow.data.nontagged.serde.AInt64SerializerDeserializer;
+import org.apache.asterix.dataflow.data.nontagged.serde.AInt8SerializerDeserializer;
+import org.apache.asterix.formats.nontagged.AqlBinaryComparatorFactoryProvider;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.primitive.ByteArrayPointable;
+import org.apache.hyracks.data.std.primitive.FloatPointable;
+import org.apache.hyracks.data.std.primitive.IntegerPointable;
+
+public class ComparisonHelper implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    private final IBinaryComparator strBinaryComp = AqlBinaryComparatorFactoryProvider.UTF8STRING_POINTABLE_INSTANCE
+            .createBinaryComparator();
+    private final IBinaryComparator circleBinaryComp = ACirclePartialBinaryComparatorFactory.INSTANCE
+            .createBinaryComparator();
+    private final IBinaryComparator durationBinaryComp = ADurationPartialBinaryComparatorFactory.INSTANCE
+            .createBinaryComparator();
+    private final IBinaryComparator intervalBinaryComp = AIntervalAscPartialBinaryComparatorFactory.INSTANCE
+            .createBinaryComparator();
+    private final IBinaryComparator lineBinaryComparator = ALinePartialBinaryComparatorFactory.INSTANCE
+            .createBinaryComparator();
+    private final IBinaryComparator pointBinaryComparator = APointPartialBinaryComparatorFactory.INSTANCE
+            .createBinaryComparator();
+    private final IBinaryComparator point3DBinaryComparator = APoint3DPartialBinaryComparatorFactory.INSTANCE
+            .createBinaryComparator();
+    private final IBinaryComparator polygonBinaryComparator = APolygonPartialBinaryComparatorFactory.INSTANCE
+            .createBinaryComparator();
+    private final IBinaryComparator rectangleBinaryComparator = ARectanglePartialBinaryComparatorFactory.INSTANCE
+            .createBinaryComparator();
+    private final IBinaryComparator uuidBinaryComparator = AUUIDPartialBinaryComparatorFactory.INSTANCE
+            .createBinaryComparator();
+    private final IBinaryComparator byteArrayComparator = new PointableBinaryComparatorFactory(
+            ByteArrayPointable.FACTORY).createBinaryComparator();
+
+    public int compare(ATypeTag typeTag1, ATypeTag typeTag2, IPointable arg1, IPointable arg2)
+            throws AlgebricksException {
+        switch (typeTag1) {
+            case INT8:
+                return compareInt8WithArg(typeTag2, arg1, arg2);
+            case INT16:
+                return compareInt16WithArg(typeTag2, arg1, arg2);
+            case INT32:
+                return compareInt32WithArg(typeTag2, arg1, arg2);
+            case INT64:
+                return compareInt64WithArg(typeTag2, arg1, arg2);
+            case FLOAT:
+                return compareFloatWithArg(typeTag2, arg1, arg2);
+            case DOUBLE:
+                return compareDoubleWithArg(typeTag2, arg1, arg2);
+            case STRING:
+                return compareStringWithArg(typeTag2, arg1, arg2);
+            case BOOLEAN:
+                return compareBooleanWithArg(typeTag2, arg1, arg2);
+            default:
+                return compareStrongTypedWithArg(typeTag1, typeTag2, arg1, arg2);
+        }
+    }
+
+    private int compareStrongTypedWithArg(ATypeTag expectedTypeTag, ATypeTag actualTypeTag, IPointable arg1,
+            IPointable arg2) throws AlgebricksException {
+        if (expectedTypeTag != actualTypeTag) {
+            throw new AlgebricksException(
+                    "Comparison is undefined between " + expectedTypeTag + " and " + actualTypeTag + ".");
+        }
+        int result = 0;
+        byte[] leftBytes = arg1.getByteArray();
+        int leftOffset = arg1.getStartOffset();
+        int leftLen = arg1.getLength() - 1;
+        byte[] rightBytes = arg2.getByteArray();
+        int rightOffset = arg2.getStartOffset();
+        int rightLen = arg2.getLength() - 1;
+
+        try {
+            switch (actualTypeTag) {
+                case YEARMONTHDURATION:
+                case TIME:
+                case DATE:
+                    result = Integer.compare(AInt32SerializerDeserializer.getInt(leftBytes, leftOffset),
+                            AInt32SerializerDeserializer.getInt(rightBytes, rightOffset));
+                    break;
+                case DAYTIMEDURATION:
+                case DATETIME:
+                    result = Long.compare(AInt64SerializerDeserializer.getLong(leftBytes, leftOffset),
+                            AInt64SerializerDeserializer.getLong(rightBytes, rightOffset));
+                    break;
+                case CIRCLE:
+                    result = circleBinaryComp.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
+                            rightLen);
+                    break;
+                case LINE:
+                    result = lineBinaryComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
+                            rightLen);
+                    break;
+                case POINT:
+                    result = pointBinaryComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
+                            rightLen);
+                    break;
+                case POINT3D:
+                    result = point3DBinaryComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
+                            rightLen);
+                    break;
+                case POLYGON:
+                    result = polygonBinaryComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
+                            rightLen);
+                    break;
+                case DURATION:
+                    result = durationBinaryComp.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
+                            rightLen);
+                    break;
+                case INTERVAL:
+                    result = intervalBinaryComp.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
+                            rightLen);
+                    break;
+                case RECTANGLE:
+                    result = rectangleBinaryComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
+                            rightLen);
+                    break;
+                case BINARY:
+                    result = byteArrayComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
+                            rightLen);
+                    break;
+                case UUID:
+                    result = uuidBinaryComparator.compare(leftBytes, leftOffset, leftLen, rightBytes, rightOffset,
+                            rightLen);
+                    break;
+                default:
+                    throw new AlgebricksException("Comparison for " + actualTypeTag + " is not supported.");
+            }
+        } catch (HyracksDataException e) {
+            throw new AlgebricksException(e);
+        }
+        return result;
+    }
+
+    private int compareBooleanWithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2) throws AlgebricksException {
+        if (typeTag2 == ATypeTag.BOOLEAN) {
+            byte b0 = arg1.getByteArray()[arg1.getStartOffset()];
+            byte b1 = arg2.getByteArray()[arg2.getStartOffset()];
+            return compareByte(b0, b1);
+        }
+        throw new AlgebricksException("Comparison is undefined between types ABoolean and " + typeTag2 + " .");
+    }
+
+    private int compareStringWithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2) throws AlgebricksException {
+        if (typeTag2 == ATypeTag.STRING) {
+            int result;
+            try {
+                result = strBinaryComp.compare(arg1.getByteArray(), arg1.getStartOffset(), arg1.getLength() - 1,
+                        arg2.getByteArray(), arg2.getStartOffset(), arg2.getLength() - 1);
+            } catch (HyracksDataException e) {
+                throw new AlgebricksException(e);
+            }
+            return result;
+        }
+        throw new AlgebricksException("Comparison is undefined between types AString and " + typeTag2 + " .");
+    }
+
+    private int compareDoubleWithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2) throws AlgebricksException {
+        byte[] leftBytes = arg1.getByteArray();
+        int leftOffset = arg1.getStartOffset();
+        byte[] rightBytes = arg2.getByteArray();
+        int rightOffset = arg2.getStartOffset();
+
+        double s = ADoubleSerializerDeserializer.getDouble(leftBytes, leftOffset);
+        switch (typeTag2) {
+            case INT8: {
+                byte v2 = AInt8SerializerDeserializer.getByte(rightBytes, rightOffset);
+                return compareDouble(s, v2);
+            }
+            case INT16: {
+                short v2 = AInt16SerializerDeserializer.getShort(rightBytes, rightOffset);
+                return compareDouble(s, v2);
+            }
+            case INT32: {
+                int v2 = AInt32SerializerDeserializer.getInt(rightBytes, rightOffset);
+                return compareDouble(s, v2);
+            }
+            case INT64: {
+                long v2 = AInt64SerializerDeserializer.getLong(rightBytes, rightOffset);
+                return compareDouble(s, v2);
+            }
+            case FLOAT: {
+                float v2 = AFloatSerializerDeserializer.getFloat(rightBytes, rightOffset);
+                return compareDouble(s, v2);
+            }
+            case DOUBLE: {
+                double v2 = ADoubleSerializerDeserializer.getDouble(rightBytes, rightOffset);
+                return compareDouble(s, v2);
+            }
+            default: {
+                throw new AlgebricksException("Comparison is undefined between types ADouble and " + typeTag2 + " .");
+            }
+        }
+    }
+
+    private int compareFloatWithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2) throws AlgebricksException {
+        byte[] leftBytes = arg1.getByteArray();
+        int leftOffset = arg1.getStartOffset();
+        byte[] rightBytes = arg2.getByteArray();
+        int rightOffset = arg2.getStartOffset();
+
+        float s = FloatPointable.getFloat(leftBytes, leftOffset);
+        switch (typeTag2) {
+            case INT8: {
+                byte v2 = AInt8SerializerDeserializer.getByte(rightBytes, rightOffset);
+                return compareFloat(s, v2);
+            }
+            case INT16: {
+                short v2 = AInt16SerializerDeserializer.getShort(rightBytes, rightOffset);
+                return compareFloat(s, v2);
+            }
+            case INT32: {
+                int v2 = AInt32SerializerDeserializer.getInt(rightBytes, rightOffset);
+                return compareFloat(s, v2);
+            }
+            case INT64: {
+                long v2 = AInt64SerializerDeserializer.getLong(rightBytes, rightOffset);
+                return compareFloat(s, v2);
+            }
+            case FLOAT: {
+                float v2 = AFloatSerializerDeserializer.getFloat(rightBytes, rightOffset);
+                return compareFloat(s, v2);
+            }
+            case DOUBLE: {
+                double v2 = ADoubleSerializerDeserializer.getDouble(rightBytes, rightOffset);
+                return compareDouble(s, v2);
+            }
+            default: {
+                throw new AlgebricksException("Comparison is undefined between types AFloat and " + typeTag2 + " .");
+            }
+        }
+    }
+
+    private int compareInt64WithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2) throws AlgebricksException {
+        byte[] leftBytes = arg1.getByteArray();
+        int leftOffset = arg1.getStartOffset();
+        byte[] rightBytes = arg2.getByteArray();
+        int rightOffset = arg2.getStartOffset();
+
+        long s = AInt64SerializerDeserializer.getLong(leftBytes, leftOffset);
+        switch (typeTag2) {
+            case INT8: {
+                byte v2 = AInt8SerializerDeserializer.getByte(rightBytes, rightOffset);
+                return compareLong(s, v2);
+            }
+            case INT16: {
+                short v2 = AInt16SerializerDeserializer.getShort(rightBytes, rightOffset);
+                return compareLong(s, v2);
+            }
+            case INT32: {
+                int v2 = AInt32SerializerDeserializer.getInt(rightBytes, rightOffset);
+                return compareLong(s, v2);
+            }
+            case INT64: {
+                long v2 = AInt64SerializerDeserializer.getLong(rightBytes, rightOffset);
+                return compareLong(s, v2);
+            }
+            case FLOAT: {
+                float v2 = AFloatSerializerDeserializer.getFloat(rightBytes, rightOffset);
+                return compareFloat(s, v2);
+            }
+            case DOUBLE: {
+                double v2 = ADoubleSerializerDeserializer.getDouble(rightBytes, rightOffset);
+                return compareDouble(s, v2);
+            }
+            default: {
+                throw new AlgebricksException("Comparison is undefined between types AInt64 and " + typeTag2 + " .");
+            }
+        }
+    }
+
+    private int compareInt32WithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2) throws AlgebricksException {
+        byte[] leftBytes = arg1.getByteArray();
+        int leftOffset = arg1.getStartOffset();
+        byte[] rightBytes = arg2.getByteArray();
+        int rightOffset = arg2.getStartOffset();
+
+        int s = IntegerPointable.getInteger(leftBytes, leftOffset);
+        switch (typeTag2) {
+            case INT8: {
+                byte v2 = AInt8SerializerDeserializer.getByte(rightBytes, rightOffset);
+                return compareInt(s, v2);
+            }
+            case INT16: {
+                short v2 = AInt16SerializerDeserializer.getShort(rightBytes, rightOffset);
+                return compareInt(s, v2);
+            }
+            case INT32: {
+                int v2 = AInt32SerializerDeserializer.getInt(rightBytes, rightOffset);
+                return compareInt(s, v2);
+            }
+            case INT64: {
+                long v2 = AInt64SerializerDeserializer.getLong(rightBytes, rightOffset);
+                return compareLong(s, v2);
+            }
+            case FLOAT: {
+                float v2 = AFloatSerializerDeserializer.getFloat(rightBytes, rightOffset);
+                return compareFloat(s, v2);
+            }
+            case DOUBLE: {
+                double v2 = ADoubleSerializerDeserializer.getDouble(rightBytes, rightOffset);
+                return compareDouble(s, v2);
+            }
+            default: {
+                throw new AlgebricksException("Comparison is undefined between types AInt32 and " + typeTag2 + " .");
+            }
+        }
+    }
+
+    private int compareInt16WithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2) throws AlgebricksException {
+        byte[] leftBytes = arg1.getByteArray();
+        int leftOffset = arg1.getStartOffset();
+        byte[] rightBytes = arg2.getByteArray();
+        int rightOffset = arg2.getStartOffset();
+
+        short s = AInt16SerializerDeserializer.getShort(leftBytes, leftOffset);
+        switch (typeTag2) {
+            case INT8: {
+                byte v2 = AInt8SerializerDeserializer.getByte(rightBytes, rightOffset);
+                return compareShort(s, v2);
+            }
+            case INT16: {
+                short v2 = AInt16SerializerDeserializer.getShort(rightBytes, rightOffset);
+                return compareShort(s, v2);
+            }
+            case INT32: {
+                int v2 = AInt32SerializerDeserializer.getInt(rightBytes, rightOffset);
+                return compareInt(s, v2);
+            }
+            case INT64: {
+                long v2 = AInt64SerializerDeserializer.getLong(rightBytes, rightOffset);
+                return compareLong(s, v2);
+            }
+            case FLOAT: {
+                float v2 = AFloatSerializerDeserializer.getFloat(rightBytes, rightOffset);
+                return compareFloat(s, v2);
+            }
+            case DOUBLE: {
+                double v2 = ADoubleSerializerDeserializer.getDouble(rightBytes, rightOffset);
+                return compareDouble(s, v2);
+            }
+            default: {
+                throw new AlgebricksException("Comparison is undefined between types AInt16 and " + typeTag2 + " .");
+            }
+        }
+    }
+
+    private int compareInt8WithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2) throws AlgebricksException {
+        byte[] leftBytes = arg1.getByteArray();
+        int leftStart = arg1.getStartOffset();
+        byte[] rightBytes = arg2.getByteArray();
+        int rightStart = arg2.getStartOffset();
+
+        byte s = AInt8SerializerDeserializer.getByte(leftBytes, leftStart);
+        switch (typeTag2) {
+            case INT8: {
+                byte v2 = AInt8SerializerDeserializer.getByte(rightBytes, rightStart);
+                return compareByte(s, v2);
+            }
+            case INT16: {
+                short v2 = AInt16SerializerDeserializer.getShort(rightBytes, rightStart);
+                return compareShort(s, v2);
+            }
+            case INT32: {
+                int v2 = AInt32SerializerDeserializer.getInt(rightBytes, rightStart);
+                return compareInt(s, v2);
+            }
+            case INT64: {
+                long v2 = AInt64SerializerDeserializer.getLong(rightBytes, rightStart);
+                return compareLong(s, v2);
+            }
+            case FLOAT: {
+                float v2 = AFloatSerializerDeserializer.getFloat(rightBytes, rightStart);
+                return compareFloat(s, v2);
+            }
+            case DOUBLE: {
+                double v2 = ADoubleSerializerDeserializer.getDouble(rightBytes, rightStart);
+                return compareDouble(s, v2);
+            }
+            default: {
+                throw new AlgebricksException("Comparison is undefined between types AInt16 and " + typeTag2 + " .");
+            }
+        }
+    }
+
+    private final int compareByte(int v1, int v2) {
+        if (v1 == v2) {
+            return 0;
+        } else if (v1 < v2) {
+            return -1;
+        } else {
+            return 1;
+        }
+    }
+
+    private final int compareShort(int v1, int v2) {
+        if (v1 == v2) {
+            return 0;
+        } else if (v1 < v2) {
+            return -1;
+        } else {
+            return 1;
+        }
+    }
+
+    private final int compareInt(int v1, int v2) {
+        if (v1 == v2) {
+            return 0;
+        } else if (v1 < v2) {
+            return -1;
+        } else {
+            return 1;
+        }
+    }
+
+    private final int compareLong(long v1, long v2) {
+        if (v1 == v2) {
+            return 0;
+        } else if (v1 < v2) {
+            return -1;
+        } else {
+            return 1;
+        }
+    }
+
+    private final int compareFloat(float v1, float v2) {
+        if (v1 == v2) {
+            return 0;
+        } else if (v1 < v2) {
+            return -1;
+        } else {
+            return 1;
+        }
+    }
+
+    private final int compareDouble(double v1, double v2) {
+        if (v1 == v2) {
+            return 0;
+        } else if (v1 < v2) {
+            return -1;
+        } else {
+            return 1;
+        }
+    }
+
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/AbstractIntervalLogicFuncDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/AbstractIntervalLogicFuncDescriptor.java
index e48e5b6..1cd3919 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/AbstractIntervalLogicFuncDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/AbstractIntervalLogicFuncDescriptor.java
@@ -20,10 +20,10 @@
 
 import java.io.DataOutput;
 
-import org.apache.asterix.dataflow.data.nontagged.serde.AIntervalSerializerDeserializer;
 import org.apache.asterix.formats.nontagged.AqlSerializerDeserializerProvider;
 import org.apache.asterix.om.base.ABoolean;
 import org.apache.asterix.om.base.ANull;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.asterix.om.types.BuiltinType;
 import org.apache.asterix.om.types.EnumDeserializer;
@@ -35,7 +35,7 @@
 import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.data.std.api.IPointable;
-import org.apache.hyracks.data.std.primitive.VoidPointable;
+import org.apache.hyracks.data.std.primitive.TaggedValuePointable;
 import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
 import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
 
@@ -57,10 +57,17 @@
 
                 return new IScalarEvaluator() {
 
+                    protected final IntervalLogic il = new IntervalLogic();
                     private ArrayBackedValueStorage resultStorage = new ArrayBackedValueStorage();
                     private DataOutput out = resultStorage.getDataOutput();
-                    private IPointable argPtr0 = new VoidPointable();
-                    private IPointable argPtr1 = new VoidPointable();
+                    private TaggedValuePointable argPtr0 = (TaggedValuePointable) TaggedValuePointable.FACTORY
+                            .createPointable();
+                    private TaggedValuePointable argPtr1 = (TaggedValuePointable) TaggedValuePointable.FACTORY
+                            .createPointable();
+                    private AIntervalPointable interval0 = (AIntervalPointable) AIntervalPointable.FACTORY
+                            .createPointable();
+                    private AIntervalPointable interval1 = (AIntervalPointable) AIntervalPointable.FACTORY
+                            .createPointable();
                     private IScalarEvaluator eval0 = args[0].createScalarEvaluator(ctx);
                     private IScalarEvaluator eval1 = args[1].createScalarEvaluator(ctx);
 
@@ -78,40 +85,31 @@
                         eval0.evaluate(tuple, argPtr0);
                         eval1.evaluate(tuple, argPtr1);
 
-                        byte[] bytes0 = argPtr0.getByteArray();
-                        int offset0 = argPtr0.getStartOffset();
-                        byte[] bytes1 = argPtr1.getByteArray();
-                        int offset1 = argPtr1.getStartOffset();
-
                         try {
-                            if (bytes0[offset0] == ATypeTag.SERIALIZED_NULL_TYPE_TAG
-                                    || bytes1[offset1] == ATypeTag.SERIALIZED_NULL_TYPE_TAG) {
+                            if (argPtr0.getTag() == ATypeTag.SERIALIZED_NULL_TYPE_TAG
+                                    || argPtr1.getTag() == ATypeTag.SERIALIZED_NULL_TYPE_TAG) {
                                 nullSerde.serialize(ANull.NULL, out);
                                 result.set(resultStorage);
                                 return;
                             }
 
-                            if (bytes0[offset0] != ATypeTag.SERIALIZED_INTERVAL_TYPE_TAG
-                                    || bytes1[offset1] != ATypeTag.SERIALIZED_INTERVAL_TYPE_TAG) {
+                            if (argPtr0.getTag() != ATypeTag.SERIALIZED_INTERVAL_TYPE_TAG
+                                    || argPtr1.getTag() != ATypeTag.SERIALIZED_INTERVAL_TYPE_TAG) {
                                 throw new AlgebricksException(getIdentifier().getName()
                                         + ": expects input type (INTERVAL, INTERVAL) but got ("
-                                        + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(bytes0[offset0]) + ", "
-                                        + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(bytes1[offset1]) + ")");
+                                        + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argPtr0.getTag()) + ", "
+                                        + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(argPtr1.getTag()) + ")");
                             }
 
-                            if (AIntervalSerializerDeserializer.getIntervalTimeType(bytes0,
-                                    offset0 + 1) != AIntervalSerializerDeserializer.getIntervalTimeType(bytes1,
-                                            offset1 + 1)) {
+                            argPtr0.getValue(interval0);
+                            argPtr1.getValue(interval1);
+
+                            if (interval0.getType() != interval1.getType()) {
                                 throw new AlgebricksException(getIdentifier().getName()
                                         + ": failed to compare intervals with different internal time type.");
                             }
 
-                            ABoolean res = (compareIntervals(
-                                    AIntervalSerializerDeserializer.getIntervalStart(bytes0, offset0 + 1),
-                                    AIntervalSerializerDeserializer.getIntervalEnd(bytes0, offset0 + 1),
-                                    AIntervalSerializerDeserializer.getIntervalStart(bytes1, offset1 + 1),
-                                    AIntervalSerializerDeserializer.getIntervalEnd(bytes1, offset1 + 1)))
-                                            ? ABoolean.TRUE : ABoolean.FALSE;
+                            ABoolean res = (compareIntervals(il, interval0, interval1)) ? ABoolean.TRUE : ABoolean.FALSE;
 
                             booleanSerde.serialize(res, out);
                         } catch (HyracksDataException hex) {
@@ -124,6 +122,7 @@
         };
     }
 
-    protected abstract boolean compareIntervals(long s1, long e1, long s2, long e2);
+    protected abstract boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException;
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/GetOverlappingIntervalDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/GetOverlappingIntervalDescriptor.java
index 94ed860..9726058 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/GetOverlappingIntervalDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/GetOverlappingIntervalDescriptor.java
@@ -20,7 +20,6 @@
 
 import java.io.DataOutput;
 
-import org.apache.asterix.dataflow.data.nontagged.serde.AIntervalSerializerDeserializer;
 import org.apache.asterix.formats.nontagged.AqlSerializerDeserializerProvider;
 import org.apache.asterix.om.base.AInterval;
 import org.apache.asterix.om.base.AMutableInterval;
@@ -28,6 +27,7 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.asterix.om.types.BuiltinType;
 import org.apache.asterix.om.types.EnumDeserializer;
@@ -40,7 +40,7 @@
 import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.data.std.api.IPointable;
-import org.apache.hyracks.data.std.primitive.VoidPointable;
+import org.apache.hyracks.data.std.primitive.TaggedValuePointable;
 import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
 import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
 
@@ -64,10 +64,17 @@
             public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws AlgebricksException {
                 return new IScalarEvaluator() {
 
+                    protected final IntervalLogic il = new IntervalLogic();
                     private ArrayBackedValueStorage resultStorage = new ArrayBackedValueStorage();
                     private DataOutput out = resultStorage.getDataOutput();
-                    private IPointable argPtr0 = new VoidPointable();
-                    private IPointable argPtr1 = new VoidPointable();
+                    private TaggedValuePointable argPtr0 = (TaggedValuePointable) TaggedValuePointable.FACTORY
+                            .createPointable();
+                    private TaggedValuePointable argPtr1 = (TaggedValuePointable) TaggedValuePointable.FACTORY
+                            .createPointable();
+                    private AIntervalPointable interval0 = (AIntervalPointable) AIntervalPointable.FACTORY
+                            .createPointable();
+                    private AIntervalPointable interval1 = (AIntervalPointable) AIntervalPointable.FACTORY
+                            .createPointable();
                     private IScalarEvaluator eval0 = args[0].createScalarEvaluator(ctx);
                     private IScalarEvaluator eval1 = args[1].createScalarEvaluator(ctx);
 
@@ -85,39 +92,30 @@
                         resultStorage.reset();
                         eval0.evaluate(tuple, argPtr0);
                         eval1.evaluate(tuple, argPtr1);
-
-                        byte[] bytes0 = argPtr0.getByteArray();
-                        int offset0 = argPtr0.getStartOffset();
-                        byte[] bytes1 = argPtr1.getByteArray();
-                        int offset1 = argPtr1.getStartOffset();
+                        byte type0 = argPtr0.getTag();
+                        byte type1 = argPtr1.getTag();
 
                         try {
-                            if (bytes0[offset0] == ATypeTag.SERIALIZED_NULL_TYPE_TAG
-                                    || bytes1[offset1] == ATypeTag.SERIALIZED_NULL_TYPE_TAG) {
+                            if (type0 == ATypeTag.SERIALIZED_NULL_TYPE_TAG
+                                    || type1 == ATypeTag.SERIALIZED_NULL_TYPE_TAG) {
                                 nullSerde.serialize(ANull.NULL, out);
-                            } else if (bytes0[offset0] == ATypeTag.SERIALIZED_INTERVAL_TYPE_TAG
-                                    && bytes0[offset0] == bytes1[offset1]) {
-                                byte type0 = AIntervalSerializerDeserializer.getIntervalTimeType(bytes0, offset0 + 1);
-                                byte type1 = AIntervalSerializerDeserializer.getIntervalTimeType(bytes1, offset1 + 1);
-                                if (type0 != type1) {
+                            } else if (type0 == ATypeTag.SERIALIZED_INTERVAL_TYPE_TAG && type0 == type1) {
+                                argPtr0.getValue(interval0);
+                                argPtr1.getValue(interval1);
+                                byte intervalType0 = interval0.getType();
+                                byte intervalType1 = interval1.getType();
+
+                                if (intervalType0 != intervalType1) {
                                     throw new AlgebricksException(getIdentifier().getName()
                                             + ": expecting two (nullable) interval values with the same internal time type but got interval of "
-                                            + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(type0)
-                                            + " and interval of "
-                                            + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(type1));
+                                            + interval0.getTypeTag() + " and interval of " + interval1.getTypeTag());
                                 }
 
-                                long start0 = AIntervalSerializerDeserializer.getIntervalStart(bytes0, offset0 + 1);
-                                long end0 = AIntervalSerializerDeserializer.getIntervalEnd(bytes0, offset0 + 1);
-
-                                long start1 = AIntervalSerializerDeserializer.getIntervalStart(bytes1, offset1 + 1);
-                                long end1 = AIntervalSerializerDeserializer.getIntervalEnd(bytes1, offset1 + 1);
-
-                                if (IntervalLogic.overlap(start0, end0, start1, end1)
-                                        || IntervalLogic.overlappedBy(start0, end0, start1, end1)
-                                        || IntervalLogic.covers(start0, end0, start1, end1)
-                                        || IntervalLogic.coveredBy(start0, end0, start1, end1)) {
-                                    aInterval.setValue(Math.max(start0, start1), Math.min(end0, end1), type0);
+                                if (il.overlap(interval0, interval1) || il.overlappedBy(interval0, interval1)
+                                        || il.covers(interval0, interval1) || il.coveredBy(interval0, interval1)) {
+                                    long start = Math.max(interval0.getStartValue(), interval1.getStartValue());
+                                    long end = Math.min(interval0.getEndValue(), interval1.getEndValue());
+                                    aInterval.setValue(start, end, intervalType0);
                                     intervalSerde.serialize(aInterval, out);
                                 } else {
                                     nullSerde.serialize(ANull.NULL, out);
@@ -125,8 +123,8 @@
                             } else {
                                 throw new AlgebricksException(getIdentifier().getName()
                                         + ": expecting two (nullable) interval values but got "
-                                        + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(bytes0[offset0]) + " and "
-                                        + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(bytes1[offset1]));
+                                        + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(type0) + " and "
+                                        + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(type1));
                             }
                         } catch (HyracksDataException hex) {
                             throw new AlgebricksException(hex);
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalAfterDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalAfterDescriptor.java
index 7513d9d..748c74f 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalAfterDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalAfterDescriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class IntervalAfterDescriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_AFTER;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,20 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_AFTER;
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.runtime.evaluators.functions.temporal.AbstractIntervalLogicFuncDescriptor#compareIntervals(long, long, long, long)
-     */
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.after(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.after(ip1, ip2);
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalBeforeDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalBeforeDescriptor.java
index 1cf6909..31189c6 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalBeforeDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalBeforeDescriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class IntervalBeforeDescriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_BEFORE;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,17 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_BEFORE;
     }
 
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.before(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.before(ip1, ip2);
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalCoveredByDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalCoveredByDescriptor.java
index b7dc767..3cb0b73 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalCoveredByDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalCoveredByDescriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class IntervalCoveredByDescriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_COVERED_BY;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,20 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_COVERED_BY;
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.runtime.evaluators.functions.temporal.AbstractIntervalLogicFuncDescriptor#compareIntervals(long, long, long, long)
-     */
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.coveredBy(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.coveredBy(ip1, ip2);
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalCoversDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalCoversDescriptor.java
index aa88033..d586885 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalCoversDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalCoversDescriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class IntervalCoversDescriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_COVERS;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,20 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_COVERS;
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.runtime.evaluators.functions.temporal.AbstractIntervalLogicFuncDescriptor#compareIntervals(long, long, long, long)
-     */
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.covers(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.covers(ip1, ip2);
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalEndedByDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalEndedByDescriptor.java
index 9149c89..1a224f7 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalEndedByDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalEndedByDescriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class IntervalEndedByDescriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_ENDED_BY;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,20 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_ENDED_BY;
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.runtime.evaluators.functions.temporal.AbstractIntervalLogicFuncDescriptor#compareIntervals(long, long, long, long)
-     */
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.endedBy(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.endedBy(ip1, ip2);
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalEndsDecriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalEndsDecriptor.java
index 64a0c42..459652e 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalEndsDecriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalEndsDecriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class IntervalEndsDecriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_ENDS;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,20 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_ENDS;
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.runtime.evaluators.functions.temporal.AbstractIntervalLogicFuncDescriptor#compareIntervals(long, long, long, long)
-     */
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.ends(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.ends(ip1, ip2);
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalLogic.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalLogic.java
index b662c5b..c8c3486 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalLogic.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalLogic.java
@@ -18,148 +18,182 @@
  */
 package org.apache.asterix.runtime.evaluators.functions.temporal;
 
-public class IntervalLogic {
+import java.io.Serializable;
 
-    public static <T extends Comparable<T>> boolean validateInterval(T s, T e) {
-        return s.compareTo(e) <= 0;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.asterix.runtime.evaluators.comparisons.ComparisonHelper;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.primitive.VoidPointable;
+
+public class IntervalLogic implements Serializable{
+
+    private static final long serialVersionUID = 1L;
+    private final ComparisonHelper ch = new ComparisonHelper();
+    private final IPointable s1 = VoidPointable.FACTORY.createPointable();
+    private final IPointable e1 = VoidPointable.FACTORY.createPointable();
+    private final IPointable s2 = VoidPointable.FACTORY.createPointable();
+    private final IPointable e2 = VoidPointable.FACTORY.createPointable();
+
+    public boolean validateInterval(AIntervalPointable ip1) throws AlgebricksException {
+        ip1.getStart(s1);
+        ip1.getEnd(e1);
+        return ch.compare(ip1.getTypeTag(), ip1.getTypeTag(), s1, e1) <= 0;
     }
 
     /**
      * Anything from interval 1 is less than anything from interval 2.
-     * <p/>
-     * |------|<br/>
-     * &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;|------|<br/>
      *
-     * @param s1
-     * @param e1
-     * @param s2
-     * @param e2
-     * @return
+     * @param ip1
+     * @param ip2
+     * @return boolean
+     * @throws AlgebricksException
+     * @see #after(AIntervalPointable, AIntervalPointable)
      */
-    public static <T extends Comparable<T>> boolean before(T s1, T e1, T s2, T e2) {
-        return e1.compareTo(s2) < 0;
+    public boolean before(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        ip1.getEnd(e1);
+        ip2.getStart(s2);
+        return ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, s2) < 0;
     }
 
-    public static <T extends Comparable<T>> boolean after(T s1, T e1, T s2, T e2) {
-        return before(s2, e2, s1, e1);
+    public boolean after(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        return before(ip2, ip1);
     }
 
     /**
      * The end of interval 1 is the same as the start of interval 2.
-     * <p/>
-     * |------|<br/>
-     * &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;|------|<br/>
      *
-     * @param s1
-     * @param e1
-     * @param s2
-     * @param e2
-     * @return
+     * @param ip1
+     * @param ip2
+     * @return boolean
+     * @throws AlgebricksException
+     * @see #metBy(AIntervalPointable, AIntervalPointable)
      */
-    public static <T extends Comparable<T>> boolean meets(T s1, T e1, T s2, T e2) {
-        return e1.compareTo(s2) == 0;
+    public boolean meets(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        ip1.getEnd(e1);
+        ip2.getStart(s2);
+        return ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, s2) == 0;
     }
 
-    public static <T extends Comparable<T>> boolean metBy(T s1, T e1, T s2, T e2) {
-        return meets(s2, e2, s1, e1);
+    public boolean metBy(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        return meets(ip2, ip1);
     }
 
     /**
      * Something at the end of interval 1 is contained as the beginning of interval 2.
-     * <p/>
-     * |------|<br/>
-     * &nbsp;&nbsp;&nbsp;&nbsp;|------|<br/>
      *
-     * @param s1
-     * @param e1
-     * @param s2
-     * @param e2
-     * @return
+     * @param ip1
+     * @param ip2
+     * @return boolean
+     * @throws AlgebricksException
+     * @see #overlappedBy(AIntervalPointable, AIntervalPointable)
      */
-    public static <T extends Comparable<T>> boolean overlaps(T s1, T e1, T s2, T e2) {
-        return s1.compareTo(s2) < 0 && e1.compareTo(s2) > 0 && e2.compareTo(e1) > 0;
+    public boolean overlaps(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        ip1.getStart(s1);
+        ip1.getEnd(e1);
+        ip2.getStart(s2);
+        ip2.getEnd(e2);
+        return ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), s1, s2) < 0
+                && ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, s2) > 0
+                && ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, e2) < 0;
     }
 
-    public static <T extends Comparable<T>> boolean overlappedBy(T s1, T e1, T s2, T e2) {
-        return overlaps(s2, e2, s1, e1);
+    public boolean overlappedBy(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        return overlaps(ip2, ip1);
     }
 
     /**
      * Something is shared by both interval 1 and interval 2.
-     * <p/>
      *
-     * @param s1
-     * @param e1
-     * @param s2
-     * @param e2
-     * @return
+     * @param ip1
+     * @param ip2
+     * @throws AlgebricksException
+     * @return boolean
      */
-    public static <T extends Comparable<T>> boolean overlap(T s1, T e1, T s2, T e2) {
-        return (s2.compareTo(s1) >= 0 && s2.compareTo(e1) < 0) || (e2.compareTo(e1) <= 0 && e2.compareTo(s1) > 0);
+    public boolean overlap(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        ip1.getStart(s1);
+        ip1.getEnd(e1);
+        ip2.getStart(s2);
+        ip2.getEnd(e2);
+        return (ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), s1, s2) <= 0
+                && ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, s2) > 0)
+                || (ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, e2) >= 0
+                        && ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), s1, e2) < 0);
     }
 
     /**
      * Anything from interval 1 is contained in the beginning of interval 2.
-     * <p/>
-     * |------|<br/>
-     * |-------|<br/>
      *
-     * @param s1
-     * @param e1
-     * @param s2
-     * @param e2
-     * @return
+     * @param ip1
+     * @param ip2
+     * @return boolean
+     * @throws AlgebricksException
+     * @see #startedBy(AIntervalPointable, AIntervalPointable)
      */
-    public static <T extends Comparable<T>> boolean starts(T s1, T e1, T s2, T e2) {
-        return s1.compareTo(s2) == 0 && e1.compareTo(e2) <= 0;
+    public boolean starts(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        ip1.getStart(s1);
+        ip1.getEnd(e1);
+        ip2.getStart(s2);
+        ip2.getEnd(e2);
+        return ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), s1, s2) == 0
+                && ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, e2) <= 0;
     }
 
-    public static <T extends Comparable<T>> boolean startedBy(T s1, T e1, T s2, T e2) {
-        return starts(s2, e2, s1, e1);
+    public boolean startedBy(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        return starts(ip2, ip1);
     }
 
     /**
      * Anything from interval 2 is in interval 1.
-     * <p/>
-     * |------|<br/>
-     * &nbsp;&nbsp;|----|<br/>
      *
-     * @param s1
-     * @param e1
-     * @param s2
-     * @param e2
-     * @return
+     * @param ip1
+     * @param ip2
+     * @return boolean
+     * @throws AlgebricksException
+     * @see #coveredBy(AIntervalPointable, AIntervalPointable)
      */
-    public static <T extends Comparable<T>> boolean covers(T s1, T e1, T s2, T e2) {
-        return s1.compareTo(s2) <= 0 && e1.compareTo(e2) >= 0;
+    public boolean covers(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        ip1.getStart(s1);
+        ip1.getEnd(e1);
+        ip2.getStart(s2);
+        ip2.getEnd(e2);
+        return ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), s1, s2) <= 0
+                && ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, e2) >= 0;
     }
 
-    public static <T extends Comparable<T>> boolean coveredBy(T s1, T e1, T s2, T e2) {
-        return covers(s2, e2, s1, e1);
+    public boolean coveredBy(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        return covers(ip2, ip1);
     }
 
     /**
      * Anything from interval 1 is from the ending part of interval 2.
-     * <p/>
-     * &nbsp;&nbsp;|-----|<br/>
-     * |------|<br/>
      *
-     * @param s1
-     * @param e1
-     * @param s2
-     * @param e2
-     * @return
+     * @param ip1
+     * @param ip2
+     * @return boolean
+     * @throws AlgebricksException
+     * @see #endedBy(AIntervalPointable, AIntervalPointable)
      */
-    public static <T extends Comparable<T>> boolean ends(T s1, T e1, T s2, T e2) {
-        return s1.compareTo(s2) >= 0 && e1.compareTo(e2) == 0;
+    public boolean ends(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        ip1.getStart(s1);
+        ip1.getEnd(e1);
+        ip2.getStart(s2);
+        ip2.getEnd(e2);
+        return ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), s1, s2) >= 0
+                && ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, e2) == 0;
     }
 
-    public static <T extends Comparable<T>> boolean endedBy(T s1, T e1, T s2, T e2) {
-        return ends(s2, e2, s1, e1);
+    public boolean endedBy(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        return ends(ip2, ip1);
     }
 
-    public static <T extends Comparable<T>> boolean equals(T s1, T e1, T s2, T e2) {
-        return s1.compareTo(s1) == 0 && e1.compareTo(e2) == 0;
+    public boolean equals(AIntervalPointable ip1, AIntervalPointable ip2) throws AlgebricksException {
+        ip1.getStart(s1);
+        ip1.getEnd(e1);
+        ip2.getStart(s2);
+        ip2.getEnd(e2);
+        return ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), s1, s2) == 0
+                && ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, e2) == 0;
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalMeetsDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalMeetsDescriptor.java
index 57ff224..288853c 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalMeetsDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalMeetsDescriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class IntervalMeetsDescriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_MEETS;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,20 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_MEETS;
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.runtime.evaluators.functions.temporal.AbstractIntervalLogicFuncDescriptor#compareIntervals(long, long, long, long)
-     */
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.meets(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.meets(ip1, ip2);
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalMetByDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalMetByDescriptor.java
index 20ef437..2985699 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalMetByDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalMetByDescriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class IntervalMetByDescriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_MET_BY;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,20 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_MET_BY;
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.runtime.evaluators.functions.temporal.AbstractIntervalLogicFuncDescriptor#compareIntervals(long, long, long, long)
-     */
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.metBy(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.metBy(ip1, ip2);
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalOverlappedByDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalOverlappedByDescriptor.java
index ced4b33..3e0a4e5 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalOverlappedByDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalOverlappedByDescriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class IntervalOverlappedByDescriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_OVERLAPPED_BY;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,20 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_OVERLAPPED_BY;
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.runtime.evaluators.functions.temporal.AbstractIntervalLogicFuncDescriptor#compareIntervals(long, long, long, long)
-     */
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.overlappedBy(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.overlappedBy(ip1, ip2);
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalOverlapsDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalOverlapsDescriptor.java
index ad79310..de5bd53 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalOverlapsDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalOverlapsDescriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class IntervalOverlapsDescriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_OVERLAPS;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,20 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_OVERLAPS;
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.runtime.evaluators.functions.temporal.AbstractIntervalLogicFuncDescriptor#compareIntervals(long, long, long, long)
-     */
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.overlaps(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.overlaps(ip1, ip2);
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalStartedByDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalStartedByDescriptor.java
index ba47b77..d609b6a 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalStartedByDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalStartedByDescriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class IntervalStartedByDescriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_STARTED_BY;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,20 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_STARTED_BY;
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.runtime.evaluators.functions.temporal.AbstractIntervalLogicFuncDescriptor#compareIntervals(long, long, long, long)
-     */
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.startedBy(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.startedBy(ip1, ip2);
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalStartsDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalStartsDescriptor.java
index 02eafdb..656c161 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalStartsDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/IntervalStartsDescriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class IntervalStartsDescriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_STARTS;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,20 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_STARTS;
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.runtime.evaluators.functions.temporal.AbstractIntervalLogicFuncDescriptor#compareIntervals(long, long, long, long)
-     */
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.starts(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.starts(ip1, ip2);
     }
 
 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/OverlapDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/OverlapDescriptor.java
index c67041b..7d3fe3f 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/OverlapDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/temporal/OverlapDescriptor.java
@@ -21,12 +21,13 @@
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.om.functions.IFunctionDescriptor;
 import org.apache.asterix.om.functions.IFunctionDescriptorFactory;
+import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 
 public class OverlapDescriptor extends AbstractIntervalLogicFuncDescriptor {
 
     private final static long serialVersionUID = 1L;
-    public final static FunctionIdentifier FID = AsterixBuiltinFunctions.INTERVAL_OVERLAPPING;
 
     public final static IFunctionDescriptorFactory FACTORY = new IFunctionDescriptorFactory() {
 
@@ -36,20 +37,15 @@
         }
     };
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.om.functions.IFunctionDescriptor#getIdentifier()
-     */
     @Override
     public FunctionIdentifier getIdentifier() {
-        return FID;
+        return AsterixBuiltinFunctions.INTERVAL_OVERLAPPING;
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.asterix.runtime.evaluators.functions.temporal.AbstractIntervalLogicFuncDescriptor#compareIntervals(long, long, long, long)
-     */
     @Override
-    protected boolean compareIntervals(long s1, long e1, long s2, long e2) {
-        return IntervalLogic.overlap(s1, e1, s2, e2);
+    protected boolean compareIntervals(IntervalLogic il, AIntervalPointable ip1, AIntervalPointable ip2)
+            throws AlgebricksException {
+        return il.overlap(ip1, ip2);
     }
 
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/ByteArrayPointable.java b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/ByteArrayPointable.java
index af54c7e..2ce5291 100644
--- a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/ByteArrayPointable.java
+++ b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/ByteArrayPointable.java
@@ -28,10 +28,10 @@
 import org.apache.hyracks.data.std.api.IHashable;
 import org.apache.hyracks.data.std.api.IPointable;
 import org.apache.hyracks.data.std.api.IPointableFactory;
-import org.apache.hyracks.data.std.api.IValueReference;
 import org.apache.hyracks.util.encoding.VarLenIntEncoderDecoder;
 
 public class ByteArrayPointable extends AbstractPointable implements IHashable, IComparable, Serializable {
+    private static final long serialVersionUID = 1L;
 
     // These three values are cached to speed up the length data access.
     // Since the we are using the variable-length encoding, we can save the repeated decoding efforts.
diff --git a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/TaggedValuePointable.java b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/TaggedValuePointable.java
new file mode 100644
index 0000000..7acd32c
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/TaggedValuePointable.java
@@ -0,0 +1,48 @@
+/*
+ * 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.hyracks.data.std.primitive;
+
+import org.apache.hyracks.api.dataflow.value.ITypeTraits;
+import org.apache.hyracks.data.std.api.AbstractPointable;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.api.IPointableFactory;
+import org.apache.hyracks.data.std.primitive.BytePointable;
+import org.apache.hyracks.data.std.primitive.VoidPointable;
+
+public class TaggedValuePointable extends AbstractPointable {
+    public static final IPointableFactory FACTORY = new IPointableFactory() {
+        private static final long serialVersionUID = 1L;
+
+        @Override
+        public ITypeTraits getTypeTraits() {
+            return VoidPointable.TYPE_TRAITS;
+        }
+
+        @Override
+        public IPointable createPointable() {
+            return new TaggedValuePointable();
+        }
+    };
+
+    public byte getTag() {
+        return BytePointable.getByte(bytes, start);
+    }
+
+    public void getValue(IPointable value) {
+        value.set(bytes, start + 1, length - 1);
+    }
+}