diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/objects/ObjectsQueries.xml b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/objects/ObjectsQueries.xml
index 150a37f..e37dee9c 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/objects/ObjectsQueries.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/objects/ObjectsQueries.xml
@@ -147,8 +147,6 @@
   <test-case FilePath="objects">
     <compilation-unit name="object_replace">
       <output-dir compare="Text">object_replace</output-dir>
-      <expected-error>function object-replace expects its 2nd input parameter to be of type primitive</expected-error>
-      <expected-error>function object-replace expects its 2nd input parameter to be of type primitive</expected-error>
       <source-location>false</source-location>
     </compilation-unit>
   </test-case>
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/objects/object_replace/object_replace.4.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/objects/object_replace/object_replace.4.query.sqlpp
index 15c0bce..9e0b525 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/objects/object_replace/object_replace.4.query.sqlpp
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/objects/object_replace/object_replace.4.query.sqlpp
@@ -19,9 +19,9 @@
 
 /*
  * Description  : Testing object_replace under different queries.
- * Expected Res : Failure
+ * Expected Res : Success
  */
 
 use TinySocial;
 
-select value object_replace({"a":1}, {"b":2}, "z");
\ No newline at end of file
+select value object_replace({"a":1, "a2": {"b":2}, "a3": {"b2":2}, "a4": {"b":2}}, {"b":2}, "z");
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/objects/object_replace/object_replace.5.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/objects/object_replace/object_replace.5.query.sqlpp
index a46a727..63008dd 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/objects/object_replace/object_replace.5.query.sqlpp
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/objects/object_replace/object_replace.5.query.sqlpp
@@ -19,9 +19,9 @@
 
 /*
  * Description  : Testing object_replace under different queries.
- * Expected Res : Failure
+ * Expected Res : Success
  */
 
 use TinySocial;
 
-select value object_replace({"a":1}, ["b","c"], "z");
\ No newline at end of file
+select value object_replace({"a":1, "a2": ["b","c"], "a3": ["c","b"], "a4": ["b","c"]}, ["b","c"], "z");
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/objects/object_replace/object_replace.4.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/objects/object_replace/object_replace.4.adm
new file mode 100644
index 0000000..b9a6bab
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/objects/object_replace/object_replace.4.adm
@@ -0,0 +1 @@
+{ "a": 1, "a2": "z", "a3": { "b2": 2 }, "a4": "z" }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/objects/object_replace/object_replace.5.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/objects/object_replace/object_replace.5.adm
new file mode 100644
index 0000000..65b79cf
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/objects/object_replace/object_replace.5.adm
@@ -0,0 +1 @@
+{ "a": 1, "a2": "z", "a3": [ "c", "b" ], "a4": "z" }
\ No newline at end of file
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/adm/AIntervalPrinterFactory.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/adm/AIntervalPrinterFactory.java
index 106b765..8cf2b88 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/adm/AIntervalPrinterFactory.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/adm/AIntervalPrinterFactory.java
@@ -34,7 +34,7 @@
     public static final IPrinter PRINTER = (byte[] b, int s, int l, PrintStream ps) -> {
         ps.print("interval(");
         byte typetag = AIntervalSerializerDeserializer.getIntervalTimeType(b, s + 1);
-        int startOffset = AIntervalSerializerDeserializer.getIntervalStartOffset(b, s + 1) - 1;
+        int startOffset = AIntervalSerializerDeserializer.getIntervalStartOffset(s + 1) - 1;
         int startSize = AIntervalSerializerDeserializer.getStartSize(b, s + 1);
         int endOffset = AIntervalSerializerDeserializer.getIntervalEndOffset(b, s + 1) - 1;
         int endSize = AIntervalSerializerDeserializer.getEndSize(b, s + 1);
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/json/clean/AIntervalPrinterFactory.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/json/clean/AIntervalPrinterFactory.java
index 67208e3..7fe13bb 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/json/clean/AIntervalPrinterFactory.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/json/clean/AIntervalPrinterFactory.java
@@ -34,7 +34,7 @@
     public static final IPrinter PRINTER = (byte[] b, int s, int l, PrintStream ps) -> {
         ps.print("{ \"interval\": { \"start\": ");
         byte typetag = AIntervalSerializerDeserializer.getIntervalTimeType(b, s + 1);
-        int startOffset = AIntervalSerializerDeserializer.getIntervalStartOffset(b, s + 1) - 1;
+        int startOffset = AIntervalSerializerDeserializer.getIntervalStartOffset(s + 1) - 1;
         int startSize = AIntervalSerializerDeserializer.getStartSize(b, s + 1);
         int endOffset = AIntervalSerializerDeserializer.getIntervalEndOffset(b, s + 1) - 1;
         int endSize = AIntervalSerializerDeserializer.getEndSize(b, s + 1);
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/json/lossless/AIntervalPrinterFactory.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/json/lossless/AIntervalPrinterFactory.java
index 9c1a4ad..04208f6 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/json/lossless/AIntervalPrinterFactory.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/json/lossless/AIntervalPrinterFactory.java
@@ -34,7 +34,7 @@
     public static final IPrinter PRINTER = (byte[] b, int s, int l, PrintStream ps) -> {
         ps.print("{ \"interval\": { \"start\": ");
         byte typetag = AIntervalSerializerDeserializer.getIntervalTimeType(b, s + 1);
-        int startOffset = AIntervalSerializerDeserializer.getIntervalStartOffset(b, s + 1) - 1;
+        int startOffset = AIntervalSerializerDeserializer.getIntervalStartOffset(s + 1) - 1;
         int startSize = AIntervalSerializerDeserializer.getStartSize(b, s + 1);
         int endOffset = AIntervalSerializerDeserializer.getIntervalEndOffset(b, s + 1) - 1;
         int endSize = AIntervalSerializerDeserializer.getEndSize(b, s + 1);
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AIntervalSerializerDeserializer.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AIntervalSerializerDeserializer.java
index 6aeeb4f..ebd7985 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AIntervalSerializerDeserializer.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AIntervalSerializerDeserializer.java
@@ -30,13 +30,14 @@
 import org.apache.hyracks.data.std.primitive.IntegerPointable;
 import org.apache.hyracks.data.std.primitive.LongPointable;
 
-/*
+/**
  * This class serializes and de-serializes the binary data representation of an interval.
  *
  * Interval {
  *   byte type;
  *   T start;
  *   T end;
+ * }
  *
  * T can be of type date, time or datetime.
  */
@@ -53,7 +54,7 @@
     public AInterval deserialize(DataInput in) throws HyracksDataException {
         try {
             byte tag = in.readByte();
-            long start = 0, end = 0;
+            long start, end;
             if (tag == ATypeTag.DATETIME.serialize()) {
                 start = in.readLong();
                 end = in.readLong();
@@ -85,28 +86,23 @@
     }
 
     public static byte getIntervalTimeType(byte[] data, int start) {
-        return BytePointable.getByte(data, getIntervalTypeOffset(data, start));
+        return BytePointable.getByte(data, start);
     }
 
-    public static int getIntervalTypeOffset(byte[] data, int start) {
-        return start;
-    }
-
-    public static int getTypeSize() {
+    private static int getTypeSize() {
         return Byte.BYTES;
     }
 
     public static long getIntervalStart(byte[] data, int start) {
         if (getIntervalTimeType(data, start) == ATypeTag.DATETIME.serialize()) {
-            return LongPointable.getLong(data, getIntervalStartOffset(data, start));
+            return LongPointable.getLong(data, getIntervalStartOffset(start));
         } else {
-            return IntegerPointable.getInteger(data, getIntervalStartOffset(data, start));
+            return IntegerPointable.getInteger(data, getIntervalStartOffset(start));
         }
     }
 
-    public static int getIntervalStartOffset(byte[] data, int start) {
-        int offset = getIntervalTypeOffset(data, start) + getTypeSize();
-        return offset;
+    public static int getIntervalStartOffset(int start) {
+        return start + getTypeSize();
     }
 
     public static int getStartSize(byte[] data, int start) {
@@ -126,8 +122,7 @@
     }
 
     public static int getIntervalEndOffset(byte[] data, int start) {
-        int offset = getIntervalStartOffset(data, start) + getStartSize(data, start);
-        return offset;
+        return getIntervalStartOffset(start) + getStartSize(data, start);
     }
 
     public static int getEndSize(byte[] data, int start) {
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
index 0063c96..0a86ffd 100644
--- 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
@@ -19,9 +19,11 @@
 
 package org.apache.asterix.om.pointables.nonvisitor;
 
+import java.io.DataOutput;
+import java.io.IOException;
+
 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.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.IJsonSerializable;
@@ -36,14 +38,16 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 
-/*
+/**
  * This class interprets the binary data representation of an interval.
  *
  * Interval {
  *   byte type;
- *   IPointable start;
- *   IPointable end;
+ *   T start;
+ *   T end;
  * }
+ *
+ * T can be of type date, time or datetime.
  */
 public class AIntervalPointable extends AbstractPointable {
 
@@ -73,84 +77,62 @@
         }
     }
 
-    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());
+        return BytePointable.getByte(bytes, start);
     }
 
     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 HyracksDataException {
-        start.set(bytes, getStartOffset(), getStartSize());
+        start.set(bytes, getIntervalStartOffset(), getStartEndSize());
     }
 
-    @Override
-    public int getStartOffset() {
-        return getTypeOffset() + getTypeSize();
-    }
-
-    public int getStartSize() throws HyracksDataException {
-        switch (getTypeTag()) {
-            case DATE:
-            case TIME:
-                return Integer.BYTES;
-            case DATETIME:
-                return Long.BYTES;
-            default:
-                throw new HyracksDataException("Unsupported interval type: " + getTypeTag() + ".");
+    public void getTaggedStart(DataOutput output) throws HyracksDataException {
+        try {
+            output.writeByte(getType());
+            output.write(bytes, getIntervalStartOffset(), getStartEndSize());
+        } catch (IOException e) {
+            throw HyracksDataException.create(e);
         }
     }
 
+    private int getIntervalStartOffset() {
+        return start + TAG_SIZE;
+    }
+
     public long getStartValue() throws HyracksDataException {
         switch (getTypeTag()) {
             case DATE:
             case TIME:
-                return IntegerPointable.getInteger(bytes, getStartOffset());
+                return IntegerPointable.getInteger(bytes, getIntervalStartOffset());
             case DATETIME:
-                return LongPointable.getLong(bytes, getStartOffset());
+                return LongPointable.getLong(bytes, getIntervalStartOffset());
             default:
                 throw new HyracksDataException("Unsupported interval type: " + getTypeTag() + ".");
         }
     }
 
-    public void getEnd(IPointable start) throws HyracksDataException {
-        start.set(bytes, getEndOffset(), getEndSize());
+    public void getEnd(IPointable end) throws HyracksDataException {
+        end.set(bytes, getEndOffset(), getStartEndSize());
     }
 
-    public int getEndOffset() throws HyracksDataException {
-        return getStartOffset() + getStartSize();
-    }
-
-    public int getEndSize() throws HyracksDataException {
-        switch (getTypeTag()) {
-            case DATE:
-            case TIME:
-                return Integer.BYTES;
-            case DATETIME:
-                return Long.BYTES;
-            default:
-                throw new HyracksDataException("Unsupported interval type: " + getTypeTag() + ".");
+    public void getTaggedEnd(DataOutput output) throws HyracksDataException {
+        try {
+            output.writeByte(getType());
+            output.write(bytes, getEndOffset(), getStartEndSize());
+        } catch (IOException e) {
+            throw HyracksDataException.create(e);
         }
     }
 
+    private int getEndOffset() throws HyracksDataException {
+        return getIntervalStartOffset() + getStartEndSize();
+    }
+
     public long getEndValue() throws HyracksDataException {
         switch (getTypeTag()) {
             case DATE:
@@ -163,4 +145,15 @@
         }
     }
 
+    private int getStartEndSize() throws HyracksDataException {
+        switch (getTypeTag()) {
+            case DATE:
+            case TIME:
+                return Integer.BYTES;
+            case DATETIME:
+                return Long.BYTES;
+            default:
+                throw new HyracksDataException("Unsupported interval type: " + getTypeTag() + ".");
+        }
+    }
 }
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
deleted file mode 100644
index 388a9d4..0000000
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/comparisons/ComparisonHelper.java
+++ /dev/null
@@ -1,478 +0,0 @@
-/*
- * 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.comparators.ComparatorUtil;
-import org.apache.asterix.dataflow.data.nontagged.comparators.LogicalComplexBinaryComparator;
-import org.apache.asterix.dataflow.data.nontagged.comparators.LogicalGenericBinaryComparator;
-import org.apache.asterix.dataflow.data.nontagged.comparators.LogicalScalarBinaryComparator;
-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.BinaryComparatorFactoryProvider;
-import org.apache.asterix.om.types.ATypeTag;
-import org.apache.asterix.om.types.EnumDeserializer;
-import org.apache.asterix.runtime.exceptions.IncompatibleTypeException;
-import org.apache.asterix.runtime.exceptions.UnsupportedTypeException;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.api.exceptions.SourceLocation;
-import org.apache.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
-import org.apache.hyracks.data.std.api.IPointable;
-import org.apache.hyracks.data.std.primitive.ByteArrayPointable;
-
-/**
- * @deprecated replaced by {@link LogicalGenericBinaryComparator}, {@link LogicalScalarBinaryComparator},
- * {@link LogicalComplexBinaryComparator}, and {@link ComparatorUtil}.
- */
-@Deprecated
-public class ComparisonHelper implements Serializable {
-    private static final long serialVersionUID = 1L;
-    static final String COMPARISON = "comparison operations (>, >=, <, and <=)";
-
-    private final IBinaryComparator strBinaryComp =
-            BinaryComparatorFactoryProvider.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();
-
-    private final SourceLocation sourceLoc;
-
-    /**
-     * @deprecated replaced by {@link LogicalGenericBinaryComparator}, {@link LogicalScalarBinaryComparator},
-     * {@link LogicalComplexBinaryComparator}, and {@link ComparatorUtil}.
-     */
-    @Deprecated
-    public ComparisonHelper(SourceLocation sourceLoc) {
-        this.sourceLoc = sourceLoc;
-    }
-
-    public int compare(ATypeTag typeTag1, ATypeTag typeTag2, IPointable arg1, IPointable arg2, Number obj1, Number obj2)
-            throws HyracksDataException {
-        switch (typeTag1) {
-            case TINYINT:
-                return compareInt8WithArg(typeTag2, arg1, arg2, obj1, obj2);
-            case SMALLINT:
-                return compareInt16WithArg(typeTag2, arg1, arg2, obj1, obj2);
-            case INTEGER:
-                return compareInt32WithArg(typeTag2, arg1, arg2, obj1, obj2);
-            case BIGINT:
-                return compareInt64WithArg(typeTag2, arg1, arg2, obj1, obj2);
-            case FLOAT:
-                return compareFloatWithArg(typeTag2, arg1, arg2, obj1, obj2);
-            case DOUBLE:
-                return compareDoubleWithArg(typeTag2, arg1, arg2, obj1, obj2);
-            case STRING:
-                return compareStringWithArg(typeTag2, arg1, arg2);
-            case BOOLEAN:
-                return compareBooleanWithArg(typeTag2, arg1, arg2);
-            default:
-                return compareStrongTypedWithArg(typeTag1, typeTag2, arg1, arg2);
-        }
-    }
-
-    public int compare(ATypeTag typeTag1, ATypeTag typeTag2, IPointable arg1, IPointable arg2)
-            throws HyracksDataException {
-        return compare(typeTag1, typeTag2, arg1, arg2, null, null);
-    }
-
-    private int compareStrongTypedWithArg(ATypeTag expectedTypeTag, ATypeTag actualTypeTag, IPointable arg1,
-            IPointable arg2) throws HyracksDataException {
-        if (expectedTypeTag != actualTypeTag) {
-            throw new IncompatibleTypeException(sourceLoc, COMPARISON, actualTypeTag.serialize(),
-                    expectedTypeTag.serialize());
-        }
-        int result;
-        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;
-
-        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 UnsupportedTypeException(sourceLoc, COMPARISON, actualTypeTag.serialize());
-        }
-        return result;
-    }
-
-    private int compareBooleanWithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2) throws HyracksDataException {
-        if (typeTag2 == ATypeTag.BOOLEAN) {
-            byte b0 = arg1.getByteArray()[arg1.getStartOffset()];
-            byte b1 = arg2.getByteArray()[arg2.getStartOffset()];
-            return compareByte(b0, b1);
-        }
-        throw new IncompatibleTypeException(sourceLoc, COMPARISON, ATypeTag.SERIALIZED_BOOLEAN_TYPE_TAG,
-                typeTag2.serialize());
-    }
-
-    private int compareStringWithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2) throws HyracksDataException {
-        if (typeTag2 == ATypeTag.STRING) {
-            return strBinaryComp.compare(arg1.getByteArray(), arg1.getStartOffset(), arg1.getLength() - 1,
-                    arg2.getByteArray(), arg2.getStartOffset(), arg2.getLength() - 1);
-        }
-        throw new IncompatibleTypeException(sourceLoc, COMPARISON, ATypeTag.SERIALIZED_STRING_TYPE_TAG,
-                typeTag2.serialize());
-    }
-
-    private int compareDoubleWithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2, Number obj1, Number obj2)
-            throws HyracksDataException {
-        byte[] leftBytes = arg1.getByteArray();
-        int leftOffset = arg1.getStartOffset();
-        byte[] rightBytes = arg2.getByteArray();
-        int rightOffset = arg2.getStartOffset();
-
-        double s = getOrDeserializeDouble(leftBytes, leftOffset, obj1);
-        switch (typeTag2) {
-            case TINYINT:
-                return compareDouble(s, getOrDeserializeTinyInt(rightBytes, rightOffset, obj2));
-            case SMALLINT:
-                return compareDouble(s, getOrDeserializeSmallInt(rightBytes, rightOffset, obj2));
-            case INTEGER:
-                return compareDouble(s, getOrDeserializeInt(rightBytes, rightOffset, obj2));
-            case BIGINT:
-                return compareDouble(s, getOrDeserializeBigInt(rightBytes, rightOffset, obj2));
-            case FLOAT:
-                return compareDouble(s, getOrDeserializeFloat(rightBytes, rightOffset, obj2));
-            case DOUBLE:
-                return compareDouble(s, getOrDeserializeDouble(rightBytes, rightOffset, obj2));
-            default: {
-                throw new IncompatibleTypeException(sourceLoc, COMPARISON, ATypeTag.SERIALIZED_DOUBLE_TYPE_TAG,
-                        typeTag2.serialize());
-            }
-        }
-    }
-
-    private int compareFloatWithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2, Number obj1, Number obj2)
-            throws HyracksDataException {
-        byte[] leftBytes = arg1.getByteArray();
-        int leftOffset = arg1.getStartOffset();
-        byte[] rightBytes = arg2.getByteArray();
-        int rightOffset = arg2.getStartOffset();
-
-        float s = getOrDeserializeFloat(leftBytes, leftOffset, obj1);
-        switch (typeTag2) {
-            case TINYINT:
-                return compareFloat(s, getOrDeserializeTinyInt(rightBytes, rightOffset, obj2));
-            case SMALLINT:
-                return compareFloat(s, getOrDeserializeSmallInt(rightBytes, rightOffset, obj2));
-            case INTEGER:
-                return compareFloat(s, getOrDeserializeInt(rightBytes, rightOffset, obj2));
-            case BIGINT:
-                return compareFloat(s, getOrDeserializeBigInt(rightBytes, rightOffset, obj2));
-            case FLOAT:
-                return compareFloat(s, getOrDeserializeFloat(rightBytes, rightOffset, obj2));
-            case DOUBLE:
-                return compareDouble(s, getOrDeserializeDouble(rightBytes, rightOffset, obj2));
-            default:
-                throw new IncompatibleTypeException(sourceLoc, COMPARISON, ATypeTag.SERIALIZED_FLOAT_TYPE_TAG,
-                        typeTag2.serialize());
-        }
-    }
-
-    private int compareInt64WithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2, Number obj1, Number obj2)
-            throws HyracksDataException {
-        byte[] leftBytes = arg1.getByteArray();
-        int leftOffset = arg1.getStartOffset();
-        byte[] rightBytes = arg2.getByteArray();
-        int rightOffset = arg2.getStartOffset();
-
-        long s = getOrDeserializeBigInt(leftBytes, leftOffset, obj1);
-        switch (typeTag2) {
-            case TINYINT:
-                return compareLong(s, getOrDeserializeTinyInt(rightBytes, rightOffset, obj2));
-            case SMALLINT:
-                return compareLong(s, getOrDeserializeSmallInt(rightBytes, rightOffset, obj2));
-            case INTEGER:
-                return compareLong(s, getOrDeserializeInt(rightBytes, rightOffset, obj2));
-            case BIGINT:
-                return compareLong(s, getOrDeserializeBigInt(rightBytes, rightOffset, obj2));
-            case FLOAT:
-                return compareFloat(s, getOrDeserializeFloat(rightBytes, rightOffset, obj2));
-            case DOUBLE:
-                return compareDouble(s, getOrDeserializeDouble(rightBytes, rightOffset, obj2));
-            default:
-                throw new IncompatibleTypeException(sourceLoc, COMPARISON, ATypeTag.SERIALIZED_INT64_TYPE_TAG,
-                        typeTag2.serialize());
-        }
-    }
-
-    private int compareInt32WithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2, Number obj1, Number obj2)
-            throws HyracksDataException {
-        byte[] leftBytes = arg1.getByteArray();
-        int leftOffset = arg1.getStartOffset();
-        byte[] rightBytes = arg2.getByteArray();
-        int rightOffset = arg2.getStartOffset();
-
-        int s = getOrDeserializeInt(leftBytes, leftOffset, obj1);
-        switch (typeTag2) {
-            case TINYINT:
-                return compareInt(s, getOrDeserializeTinyInt(rightBytes, rightOffset, obj2));
-            case SMALLINT:
-                return compareInt(s, getOrDeserializeSmallInt(rightBytes, rightOffset, obj2));
-            case INTEGER:
-                return compareInt(s, getOrDeserializeInt(rightBytes, rightOffset, obj2));
-            case BIGINT:
-                return compareLong(s, getOrDeserializeBigInt(rightBytes, rightOffset, obj2));
-            case FLOAT:
-                return compareFloat(s, getOrDeserializeFloat(rightBytes, rightOffset, obj2));
-            case DOUBLE:
-                return compareDouble(s, getOrDeserializeDouble(rightBytes, rightOffset, obj2));
-            default:
-                throw new IncompatibleTypeException(sourceLoc, COMPARISON, ATypeTag.SERIALIZED_INT32_TYPE_TAG,
-                        typeTag2.serialize());
-        }
-    }
-
-    private int compareInt16WithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2, Number obj1, Number obj2)
-            throws HyracksDataException {
-        byte[] leftBytes = arg1.getByteArray();
-        int leftOffset = arg1.getStartOffset();
-        byte[] rightBytes = arg2.getByteArray();
-        int rightOffset = arg2.getStartOffset();
-
-        short s = getOrDeserializeSmallInt(leftBytes, leftOffset, obj1);
-        switch (typeTag2) {
-            case TINYINT:
-                return compareShort(s, getOrDeserializeTinyInt(rightBytes, rightOffset, obj2));
-            case SMALLINT:
-                return compareShort(s, getOrDeserializeSmallInt(rightBytes, rightOffset, obj2));
-            case INTEGER:
-                return compareInt(s, getOrDeserializeInt(rightBytes, rightOffset, obj2));
-            case BIGINT:
-                return compareLong(s, getOrDeserializeBigInt(rightBytes, rightOffset, obj2));
-            case FLOAT:
-                return compareFloat(s, getOrDeserializeFloat(rightBytes, rightOffset, obj2));
-            case DOUBLE:
-                return compareDouble(s, getOrDeserializeDouble(rightBytes, rightOffset, obj2));
-            default:
-                throw new IncompatibleTypeException(sourceLoc, COMPARISON, ATypeTag.SERIALIZED_INT16_TYPE_TAG,
-                        typeTag2.serialize());
-        }
-    }
-
-    private int compareInt8WithArg(ATypeTag typeTag2, IPointable arg1, IPointable arg2, Number obj1, Number obj2)
-            throws HyracksDataException {
-        byte[] leftBytes = arg1.getByteArray();
-        int leftStart = arg1.getStartOffset();
-        byte[] rightBytes = arg2.getByteArray();
-        int rightStart = arg2.getStartOffset();
-
-        byte s = getOrDeserializeTinyInt(leftBytes, leftStart, obj1);
-        switch (typeTag2) {
-            case TINYINT:
-                return compareByte(s, getOrDeserializeTinyInt(rightBytes, rightStart, obj2));
-            case SMALLINT:
-                return compareShort(s, getOrDeserializeSmallInt(rightBytes, rightStart, obj2));
-            case INTEGER:
-                return compareInt(s, getOrDeserializeInt(rightBytes, rightStart, obj2));
-            case BIGINT:
-                return compareLong(s, getOrDeserializeBigInt(rightBytes, rightStart, obj2));
-            case FLOAT:
-                return compareFloat(s, getOrDeserializeFloat(rightBytes, rightStart, obj2));
-            case DOUBLE:
-                return compareDouble(s, getOrDeserializeDouble(rightBytes, rightStart, obj2));
-            default:
-                throw new IncompatibleTypeException(sourceLoc, COMPARISON, ATypeTag.SERIALIZED_INT8_TYPE_TAG,
-                        typeTag2.serialize());
-        }
-    }
-
-    private final byte getOrDeserializeTinyInt(byte[] bytes, int offset, Number obj) {
-        return obj == null ? AInt8SerializerDeserializer.getByte(bytes, offset) : obj.byteValue();
-    }
-
-    private final short getOrDeserializeSmallInt(byte[] bytes, int offset, Number obj) {
-        return obj == null ? AInt16SerializerDeserializer.getShort(bytes, offset) : obj.shortValue();
-    }
-
-    private final int getOrDeserializeInt(byte[] bytes, int offset, Number obj) {
-        return obj == null ? AInt32SerializerDeserializer.getInt(bytes, offset) : obj.intValue();
-    }
-
-    private final long getOrDeserializeBigInt(byte[] bytes, int offset, Number obj) {
-        return obj == null ? AInt64SerializerDeserializer.getLong(bytes, offset) : obj.longValue();
-    }
-
-    private final float getOrDeserializeFloat(byte[] bytes, int offset, Number obj) {
-        return obj == null ? AFloatSerializerDeserializer.getFloat(bytes, offset) : obj.floatValue();
-    }
-
-    private final double getOrDeserializeDouble(byte[] bytes, int offset, Number obj) {
-        return obj == null ? ADoubleSerializerDeserializer.getDouble(bytes, offset) : obj.doubleValue();
-    }
-
-    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) {
-        return Float.compare(v1, v2);
-    }
-
-    private final int compareDouble(double v1, double v2) {
-        return Double.compare(v1, v2);
-    }
-
-    /**
-     * When field value falls into the primitive type groups, we consider to cache its value instead of deserialize it
-     * every time.
-     *
-     * @param bytes
-     * @return primitive value as Number
-     */
-    public Number getNumberValue(byte[] bytes) {
-        ATypeTag aTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(bytes[0]);
-        int offset = 1;
-        if (aTypeTag == null) {
-            return null;
-        }
-        switch (aTypeTag) {
-            case TINYINT:
-                return AInt8SerializerDeserializer.getByte(bytes, offset);
-            case SMALLINT:
-                return AInt16SerializerDeserializer.getShort(bytes, offset);
-            case INTEGER:
-                return AInt32SerializerDeserializer.getInt(bytes, offset);
-            case BIGINT:
-                return AInt64SerializerDeserializer.getLong(bytes, offset);
-            case FLOAT:
-                return AFloatSerializerDeserializer.getFloat(bytes, offset);
-            case DOUBLE:
-                return ADoubleSerializerDeserializer.getDouble(bytes, offset);
-            default:
-                return null;
-        }
-    }
-}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/records/RecordReplaceDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/records/RecordReplaceDescriptor.java
index 03d0088..e7b642d 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/records/RecordReplaceDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/records/RecordReplaceDescriptor.java
@@ -67,7 +67,7 @@
                 for (int i = 0; i < args.length; i++) {
                     argEvals[i] = args[i].createScalarEvaluator(ctx);
                 }
-                return new RecordReplaceEvaluator(sourceLoc, argEvals[0], argEvals[1], argEvals[2], argTypes);
+                return new RecordReplaceEvaluator(argEvals[0], argEvals[1], argEvals[2], argTypes);
             }
         };
     }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/records/RecordReplaceEvaluator.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/records/RecordReplaceEvaluator.java
index 13acefb..550a84e 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/records/RecordReplaceEvaluator.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/evaluators/functions/records/RecordReplaceEvaluator.java
@@ -24,21 +24,18 @@
 import java.util.List;
 
 import org.apache.asterix.builders.RecordBuilder;
-import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.formats.nontagged.BinaryComparatorFactoryProvider;
 import org.apache.asterix.om.pointables.ARecordVisitablePointable;
 import org.apache.asterix.om.pointables.base.DefaultOpenFieldType;
 import org.apache.asterix.om.pointables.base.IVisitablePointable;
 import org.apache.asterix.om.types.ATypeTag;
 import org.apache.asterix.om.types.BuiltinType;
 import org.apache.asterix.om.types.IAType;
-import org.apache.asterix.om.types.hierachy.ATypeHierarchy;
-import org.apache.asterix.runtime.evaluators.comparisons.ComparisonHelper;
 import org.apache.asterix.runtime.evaluators.functions.CastTypeEvaluator;
 import org.apache.asterix.runtime.evaluators.functions.PointableHelper;
-import org.apache.asterix.runtime.exceptions.TypeMismatchException;
 import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.api.exceptions.SourceLocation;
 import org.apache.hyracks.data.std.api.IPointable;
 import org.apache.hyracks.data.std.primitive.VoidPointable;
 import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
@@ -52,28 +49,24 @@
     private final ArrayBackedValueStorage resultStorage = new ArrayBackedValueStorage();
     private final DataOutput resultOutput = resultStorage.getDataOutput();
     private final RecordBuilder outRecordBuilder = new RecordBuilder();
-    private final VoidPointable existingValuePtr = new VoidPointable();
-    private final VoidPointable oldValuePtr = new VoidPointable();
     private final IScalarEvaluator eval0;
     private final IScalarEvaluator eval1;
     private final IScalarEvaluator eval2;
     private final ARecordVisitablePointable openRecordPointable;
     private final CastTypeEvaluator inputRecordCaster;
     private final CastTypeEvaluator newValueRecordCaster;
-    private final SourceLocation sourceLoc;
-    // TODO(ali): switch to ILogicalBinaryComparator
-    private final ComparisonHelper comparisonHelper;
+    private final IBinaryComparator comp;
 
-    RecordReplaceEvaluator(SourceLocation sourceLoc, IScalarEvaluator eval0, IScalarEvaluator eval1,
-            IScalarEvaluator eval2, IAType[] argTypes) {
-        this.sourceLoc = sourceLoc;
+    RecordReplaceEvaluator(IScalarEvaluator eval0, IScalarEvaluator eval1, IScalarEvaluator eval2, IAType[] argTypes) {
         this.eval0 = eval0;
         this.eval1 = eval1;
         this.eval2 = eval2;
         openRecordPointable = new ARecordVisitablePointable(DefaultOpenFieldType.NESTED_OPEN_RECORD_TYPE);
         inputRecordCaster = new CastTypeEvaluator(BuiltinType.ANY, argTypes[0], eval0);
         newValueRecordCaster = new CastTypeEvaluator(BuiltinType.ANY, argTypes[2], eval2);
-        comparisonHelper = new ComparisonHelper(sourceLoc);
+        // comp compares a value existing in the input record with the provided value. the input record is casted open
+        comp = BinaryComparatorFactoryProvider.INSTANCE.getBinaryComparatorFactory(BuiltinType.ANY, argTypes[1], true)
+                .createBinaryComparator();
     }
 
     @Override
@@ -94,21 +87,17 @@
             result.set(resultStorage);
             return;
         }
-        if (oldValueType.isDerivedType()) {
-            throw new TypeMismatchException(sourceLoc, BuiltinFunctions.RECORD_REPLACE, 1, oldValueType.serialize(),
-                    "primitive");
-        }
         inputRecordCaster.evaluate(tuple, inputRecordPointable);
         final ATypeTag newValueType = PointableHelper.getTypeTag(newValuePointable);
         if (newValueType.isDerivedType()) {
             newValueRecordCaster.evaluate(tuple, newValuePointable);
         }
         resultStorage.reset();
-        buildOutputRecord(oldValueType);
+        buildOutputRecord();
         result.set(resultStorage);
     }
 
-    private void buildOutputRecord(ATypeTag oldValueTypeTag) throws HyracksDataException {
+    private void buildOutputRecord() throws HyracksDataException {
         openRecordPointable.set(inputRecordPointable);
         outRecordBuilder.reset(DefaultOpenFieldType.NESTED_OPEN_RECORD_TYPE);
         outRecordBuilder.init();
@@ -117,8 +106,7 @@
         for (int i = 0, fieldCount = fieldNames.size(); i < fieldCount; i++) {
             final IVisitablePointable fieldName = fieldNames.get(i);
             final IVisitablePointable fieldValue = fieldValues.get(i);
-            final ATypeTag existingValueTypeTag = PointableHelper.getTypeTag(fieldValue);
-            if (isEqual(existingValueTypeTag, fieldValue, oldValueTypeTag, oldValuePointable)) {
+            if (isEqual(fieldValue, oldValuePointable)) {
                 outRecordBuilder.addField(fieldName, newValuePointable);
             } else {
                 outRecordBuilder.addField(fieldName, fieldValue);
@@ -144,17 +132,8 @@
         }
     }
 
-    private boolean isEqual(ATypeTag typeTag1, IPointable value1, ATypeTag typeTag2, IPointable value2)
-            throws HyracksDataException {
-        if (!ATypeHierarchy.isCompatible(typeTag1, typeTag2)) {
-            return false;
-        }
-        setValuePointer(value1, existingValuePtr);
-        setValuePointer(value2, oldValuePtr);
-        return comparisonHelper.compare(typeTag1, typeTag2, existingValuePtr, oldValuePtr) == 0;
-    }
-
-    private static void setValuePointer(IPointable src, IPointable value) {
-        value.set(src.getByteArray(), src.getStartOffset() + 1, src.getLength() - 1);
+    private boolean isEqual(IPointable value1, IPointable value2) throws HyracksDataException {
+        return comp.compare(value1.getByteArray(), value1.getStartOffset(), value1.getLength(), value2.getByteArray(),
+                value2.getStartOffset(), value2.getLength()) == 0;
     }
 }
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 fba0c42..53993c5 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
@@ -27,7 +27,6 @@
 import org.apache.asterix.om.types.BuiltinType;
 import org.apache.asterix.runtime.evaluators.base.AbstractScalarFunctionDynamicDescriptor;
 import org.apache.asterix.runtime.evaluators.functions.PointableHelper;
-import org.apache.asterix.runtime.exceptions.IncompatibleTypeException;
 import org.apache.asterix.runtime.exceptions.TypeMismatchException;
 import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
 import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
@@ -40,6 +39,7 @@
 import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
 
 public abstract class AbstractIntervalLogicFuncDescriptor extends AbstractScalarFunctionDynamicDescriptor {
+
     private static final long serialVersionUID = 1L;
 
     @Override
@@ -52,7 +52,7 @@
 
                 return new IScalarEvaluator() {
 
-                    protected final IntervalLogic il = new IntervalLogic(sourceLoc);
+                    private final IntervalLogic il = new IntervalLogic();
                     private ArrayBackedValueStorage resultStorage = new ArrayBackedValueStorage();
                     private DataOutput out = resultStorage.getDataOutput();
                     private TaggedValuePointable argPtr0 = TaggedValuePointable.FACTORY.createPointable();
@@ -74,11 +74,9 @@
                         resultStorage.reset();
                         eval0.evaluate(tuple, argPtr0);
                         eval1.evaluate(tuple, argPtr1);
-
                         if (PointableHelper.checkAndSetMissingOrNull(result, argPtr0, argPtr1)) {
                             return;
                         }
-
                         byte typeTag0 = argPtr0.getTag();
                         if (typeTag0 != ATypeTag.SERIALIZED_INTERVAL_TYPE_TAG) {
                             throw new TypeMismatchException(sourceLoc, getIdentifier(), 0, typeTag0,
@@ -89,14 +87,8 @@
                             throw new TypeMismatchException(sourceLoc, getIdentifier(), 1, typeTag1,
                                     ATypeTag.SERIALIZED_INTERVAL_TYPE_TAG);
                         }
-
                         argPtr0.getValue(interval0);
                         argPtr1.getValue(interval1);
-
-                        if (typeTag0 != typeTag1) {
-                            throw new IncompatibleTypeException(sourceLoc, getIdentifier(), typeTag0, typeTag1);
-                        }
-
                         ABoolean res = compareIntervals(il, interval0, interval1) ? ABoolean.TRUE : ABoolean.FALSE;
                         booleanSerde.serialize(res, out);
                         result.set(resultStorage);
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 f288902..9578bad 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
@@ -65,7 +65,7 @@
             public IScalarEvaluator createScalarEvaluator(final IHyracksTaskContext ctx) throws HyracksDataException {
                 return new IScalarEvaluator() {
 
-                    protected final IntervalLogic il = new IntervalLogic(sourceLoc);
+                    private final IntervalLogic il = new IntervalLogic();
                     private ArrayBackedValueStorage resultStorage = new ArrayBackedValueStorage();
                     private DataOutput out = resultStorage.getDataOutput();
                     private TaggedValuePointable argPtr0 = TaggedValuePointable.FACTORY.createPointable();
@@ -120,7 +120,6 @@
                                 nullSerde.serialize(ANull.NULL, out);
                             }
                             result.set(resultStorage);
-                            return;
                         } else if (type0 != ATypeTag.SERIALIZED_INTERVAL_TYPE_TAG) {
                             throw new TypeMismatchException(sourceLoc, getIdentifier(), 0, type0,
                                     ATypeTag.SERIALIZED_INTERVAL_TYPE_TAG);
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 d5d23fb..2178b48 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,194 +18,204 @@
  */
 package org.apache.asterix.runtime.evaluators.functions.temporal;
 
-import java.io.Serializable;
-
+import org.apache.asterix.formats.nontagged.BinaryComparatorFactoryProvider;
 import org.apache.asterix.om.pointables.nonvisitor.AIntervalPointable;
-import org.apache.asterix.runtime.evaluators.comparisons.ComparisonHelper;
+import org.apache.asterix.om.types.BuiltinType;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.api.exceptions.SourceLocation;
-import org.apache.hyracks.data.std.api.IPointable;
-import org.apache.hyracks.data.std.primitive.VoidPointable;
+import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
 
-public class IntervalLogic implements Serializable {
+class IntervalLogic {
 
-    private static final long serialVersionUID = 1L;
-    // TODO(ali): switch to ILogicalBinaryComparator
-    private final ComparisonHelper ch;
-    private final transient IPointable s1 = VoidPointable.FACTORY.createPointable();
-    private final transient IPointable e1 = VoidPointable.FACTORY.createPointable();
-    private final transient IPointable s2 = VoidPointable.FACTORY.createPointable();
-    private final transient IPointable e2 = VoidPointable.FACTORY.createPointable();
+    private final IBinaryComparator comp;
+    private final ArrayBackedValueStorage s1 = new ArrayBackedValueStorage();
+    private final ArrayBackedValueStorage e1 = new ArrayBackedValueStorage();
+    private final ArrayBackedValueStorage s2 = new ArrayBackedValueStorage();
+    private final ArrayBackedValueStorage e2 = new ArrayBackedValueStorage();
 
-    public IntervalLogic(SourceLocation sourceLoc) {
-        ch = new ComparisonHelper(sourceLoc);
-    }
-
-    public boolean validateInterval(AIntervalPointable ip1) throws HyracksDataException {
-        ip1.getStart(s1);
-        ip1.getEnd(e1);
-        return ch.compare(ip1.getTypeTag(), ip1.getTypeTag(), s1, e1) <= 0;
+    IntervalLogic() {
+        comp = BinaryComparatorFactoryProvider.INSTANCE
+                .getBinaryComparatorFactory(BuiltinType.ANY, BuiltinType.ANY, true).createBinaryComparator();
     }
 
     /**
      * Anything from interval 1 is less than anything from interval 2.
      *
-     * @param ip1
-     * @param ip2
+     * @param ip1 interval 1
+     * @param ip2 interval 2
      * @return boolean
-     * @throws HyracksDataException
+     * @throws HyracksDataException IOException
      * @see #after(AIntervalPointable, AIntervalPointable)
      */
-    public boolean before(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
-        ip1.getEnd(e1);
-        ip2.getStart(s2);
-        return ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, s2) < 0;
+    boolean before(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+        e1.reset();
+        s2.reset();
+        ip1.getTaggedEnd(e1.getDataOutput());
+        ip2.getTaggedStart(s2.getDataOutput());
+        return comp.compare(e1.getByteArray(), e1.getStartOffset(), e1.getLength(), s2.getByteArray(),
+                s2.getStartOffset(), s2.getLength()) < 0;
     }
 
-    public boolean after(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+    boolean after(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
         return before(ip2, ip1);
     }
 
     /**
      * The end of interval 1 is the same as the start of interval 2.
      *
-     * @param ip1
-     * @param ip2
+     * @param ip1 interval 1
+     * @param ip2 interval 2
      * @return boolean
-     * @throws HyracksDataException
+     * @throws HyracksDataException IOException
      * @see #metBy(AIntervalPointable, AIntervalPointable)
      */
-    public boolean meets(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
-        ip1.getEnd(e1);
-        ip2.getStart(s2);
-        return ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, s2) == 0;
+    boolean meets(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+        e1.reset();
+        s2.reset();
+        ip1.getTaggedEnd(e1.getDataOutput());
+        ip2.getTaggedStart(s2.getDataOutput());
+        return comp.compare(e1.getByteArray(), e1.getStartOffset(), e1.getLength(), s2.getByteArray(),
+                s2.getStartOffset(), s2.getLength()) == 0;
     }
 
-    public boolean metBy(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+    boolean metBy(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
         return meets(ip2, ip1);
     }
 
     /**
      * Something at the end of interval 1 is contained as the beginning of interval 2.
      *
-     * @param ip1
-     * @param ip2
+     * @param ip1 interval 1
+     * @param ip2 interval 2
      * @return boolean
-     * @throws HyracksDataException
+     * @throws HyracksDataException IOException
      * @see #overlappedBy(AIntervalPointable, AIntervalPointable)
      */
-    public boolean overlaps(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
-        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;
+    boolean overlaps(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+        s1.reset();
+        e1.reset();
+        s2.reset();
+        e2.reset();
+        ip1.getTaggedStart(s1.getDataOutput());
+        ip1.getTaggedEnd(e1.getDataOutput());
+        ip2.getTaggedStart(s2.getDataOutput());
+        ip2.getTaggedEnd(e2.getDataOutput());
+        return comp.compare(s1.getByteArray(), s1.getStartOffset(), s1.getLength(), s2.getByteArray(),
+                s2.getStartOffset(), s2.getLength()) < 0
+                && comp.compare(e1.getByteArray(), e1.getStartOffset(), e1.getLength(), s2.getByteArray(),
+                        s2.getStartOffset(), s2.getLength()) > 0
+                && comp.compare(e1.getByteArray(), e1.getStartOffset(), e1.getLength(), e2.getByteArray(),
+                        e2.getStartOffset(), e2.getLength()) < 0;
     }
 
-    public boolean overlappedBy(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+    boolean overlappedBy(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
         return overlaps(ip2, ip1);
     }
 
     /**
      * Something is shared by both interval 1 and interval 2.
      *
-     * @param ip1
-     * @param ip2
-     * @throws HyracksDataException
+     * @param ip1 interval 1
+     * @param ip2 interval 2
+     * @throws HyracksDataException IOException
      * @return boolean
      */
-    public boolean overlapping(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
-        ip1.getStart(s1);
-        ip1.getEnd(e1);
-        ip2.getStart(s2);
-        ip2.getEnd(e2);
-        return ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), s1, e2) < 0
-                && ch.compare(ip1.getTypeTag(), ip2.getTypeTag(), e1, s2) > 0;
+    boolean overlapping(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+        s1.reset();
+        e1.reset();
+        s2.reset();
+        e2.reset();
+        ip1.getTaggedStart(s1.getDataOutput());
+        ip1.getTaggedEnd(e1.getDataOutput());
+        ip2.getTaggedStart(s2.getDataOutput());
+        ip2.getTaggedEnd(e2.getDataOutput());
+        return comp.compare(s1.getByteArray(), s1.getStartOffset(), s1.getLength(), e2.getByteArray(),
+                e2.getStartOffset(), e2.getLength()) < 0
+                && comp.compare(e1.getByteArray(), e1.getStartOffset(), e1.getLength(), s2.getByteArray(),
+                        s2.getStartOffset(), s2.getLength()) > 0;
     }
 
     /**
      * Anything from interval 1 is contained in the beginning of interval 2.
      *
-     * @param ip1
-     * @param ip2
+     * @param ip1 interval 1
+     * @param ip2 interval 2
      * @return boolean
-     * @throws HyracksDataException
+     * @throws HyracksDataException IOException
      * @see #startedBy(AIntervalPointable, AIntervalPointable)
      */
-    public boolean starts(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
-        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;
+    boolean starts(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+        s1.reset();
+        e1.reset();
+        s2.reset();
+        e2.reset();
+        ip1.getTaggedStart(s1.getDataOutput());
+        ip1.getTaggedEnd(e1.getDataOutput());
+        ip2.getTaggedStart(s2.getDataOutput());
+        ip2.getTaggedEnd(e2.getDataOutput());
+        return comp.compare(s1.getByteArray(), s1.getStartOffset(), s1.getLength(), s2.getByteArray(),
+                s2.getStartOffset(), s2.getLength()) == 0
+                && comp.compare(e1.getByteArray(), e1.getStartOffset(), e1.getLength(), e2.getByteArray(),
+                        e2.getStartOffset(), e2.getLength()) <= 0;
     }
 
-    public boolean startedBy(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+    boolean startedBy(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
         return starts(ip2, ip1);
     }
 
     /**
      * Anything from interval 2 is in interval 1.
      *
-     * @param ip1
-     * @param ip2
+     * @param ip1 interval 1
+     * @param ip2 interval 2
      * @return boolean
-     * @throws HyracksDataException
+     * @throws HyracksDataException IOException
      * @see #coveredBy(AIntervalPointable, AIntervalPointable)
      */
-    public boolean covers(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
-        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;
+    boolean covers(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+        s1.reset();
+        e1.reset();
+        s2.reset();
+        e2.reset();
+        ip1.getTaggedStart(s1.getDataOutput());
+        ip1.getTaggedEnd(e1.getDataOutput());
+        ip2.getTaggedStart(s2.getDataOutput());
+        ip2.getTaggedEnd(e2.getDataOutput());
+        return comp.compare(s1.getByteArray(), s1.getStartOffset(), s1.getLength(), s2.getByteArray(),
+                s2.getStartOffset(), s2.getLength()) <= 0
+                && comp.compare(e1.getByteArray(), e1.getStartOffset(), e1.getLength(), e2.getByteArray(),
+                        e2.getStartOffset(), e2.getLength()) >= 0;
     }
 
-    public boolean coveredBy(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+    boolean coveredBy(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
         return covers(ip2, ip1);
     }
 
     /**
      * Anything from interval 1 is from the ending part of interval 2.
      *
-     * @param ip1
-     * @param ip2
+     * @param ip1 interval 1
+     * @param ip2 interval 2
      * @return boolean
-     * @throws HyracksDataException
+     * @throws HyracksDataException IOException
      * @see #endedBy(AIntervalPointable, AIntervalPointable)
      */
-    public boolean ends(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
-        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;
+    boolean ends(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+        s1.reset();
+        e1.reset();
+        s2.reset();
+        e2.reset();
+        ip1.getTaggedStart(s1.getDataOutput());
+        ip1.getTaggedEnd(e1.getDataOutput());
+        ip2.getTaggedStart(s2.getDataOutput());
+        ip2.getTaggedEnd(e2.getDataOutput());
+        return comp.compare(s1.getByteArray(), s1.getStartOffset(), s1.getLength(), s2.getByteArray(),
+                s2.getStartOffset(), s2.getLength()) >= 0
+                && comp.compare(e1.getByteArray(), e1.getStartOffset(), e1.getLength(), e2.getByteArray(),
+                        e2.getStartOffset(), e2.getLength()) == 0;
     }
 
-    public boolean endedBy(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
+    boolean endedBy(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
         return ends(ip2, ip1);
     }
-
-    /**
-     * Intervals with the same start and end time.
-     *
-     * @param ip1
-     * @param ip2
-     * @return boolean
-     * @throws HyracksDataException
-     */
-    public boolean same(AIntervalPointable ip1, AIntervalPointable ip2) throws HyracksDataException {
-        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;
-    }
-
 }