diff --git a/velox/functions/lib/SIMDComparisonUtil.h b/velox/functions/lib/SIMDComparisonUtil.h
new file mode 100644
index 000000000000..287fc8c56212
--- /dev/null
+++ b/velox/functions/lib/SIMDComparisonUtil.h
@@ -0,0 +1,302 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License 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.
+ */
+#pragma once
+
+#include "velox/expression/VectorFunction.h"
+
+namespace facebook::velox::functions {
+
+namespace detail {
+
+template <typename T, bool kIsConstant>
+inline auto loadSimdData(const T* rawData, vector_size_t offset) {
+  using d_type = xsimd::batch<T>;
+  if constexpr (kIsConstant) {
+    return xsimd::broadcast<T>(rawData[0]);
+  }
+  return d_type::load_unaligned(rawData + offset);
+}
+
+inline uint64_t to64Bits(const int8_t* resultData) {
+  using d_type = xsimd::batch<int8_t>;
+  constexpr auto numScalarElements = d_type::size;
+  static_assert(
+      numScalarElements == 16 || numScalarElements == 32 ||
+          numScalarElements == 64,
+      "Unsupported number of scalar elements");
+  uint64_t res = 0UL;
+  if constexpr (numScalarElements == 64) {
+    res = simd::toBitMask(
+        xsimd::batch_bool<int8_t>(d_type::load_unaligned(resultData)));
+  } else if constexpr (numScalarElements == 32) {
+    auto* addr = reinterpret_cast<uint32_t*>(&res);
+    *(addr) = simd::toBitMask(
+        xsimd::batch_bool<int8_t>(d_type::load_unaligned(resultData)));
+    *(addr + 1) = simd::toBitMask(
+        xsimd::batch_bool<int8_t>(d_type::load_unaligned(resultData + 32)));
+  } else if constexpr (numScalarElements == 16) {
+    auto* addr = reinterpret_cast<uint16_t*>(&res);
+    *(addr) = simd::toBitMask(
+        xsimd::batch_bool<int8_t>(d_type::load_unaligned(resultData)));
+    *(addr + 1) = simd::toBitMask(
+        xsimd::batch_bool<int8_t>(d_type::load_unaligned(resultData + 16)));
+    *(addr + 2) = simd::toBitMask(
+        xsimd::batch_bool<int8_t>(d_type::load_unaligned(resultData + 32)));
+    *(addr + 3) = simd::toBitMask(
+        xsimd::batch_bool<int8_t>(d_type::load_unaligned(resultData + 48)));
+  }
+  return res;
+}
+
+template <typename A, typename B, typename Compare>
+void applyAutoSimdComparisonInternal(
+    const SelectivityVector& rows,
+    const A* __restrict rawA,
+    const B* __restrict rawB,
+    Compare cmp,
+    VectorPtr& result) {
+  int8_t tempBuffer[64];
+  int8_t* __restrict resultData = tempBuffer;
+  const vector_size_t rowsBegin = rows.begin();
+  const vector_size_t rowsEnd = rows.end();
+  auto* rowsData = reinterpret_cast<const uint64_t*>(rows.allBits());
+  auto* resultVector = result->asUnchecked<FlatVector<bool>>();
+  auto* rawResult = resultVector->mutableRawValues<uint64_t>();
+  if (rows.isAllSelected()) {
+    auto i = 0;
+    for (; i + 64 <= rowsEnd; i += 64) {
+      for (auto j = 0; j < 64; ++j) {
+        resultData[j] = cmp(rawA, rawB, i + j) ? -1 : 0;
+      }
+      rawResult[i / 64] = to64Bits(resultData);
+    }
+    for (; i < rowsEnd; ++i) {
+      bits::setBit(rawResult, i, cmp(rawA, rawB, i));
+    }
+  } else {
+    static constexpr uint64_t kAllSet = -1ULL;
+    bits::forEachWord(
+        rowsBegin,
+        rowsEnd,
+        [&](int32_t idx, uint64_t mask) {
+          auto word = rowsData[idx] & mask;
+          if (!word) {
+            return;
+          }
+          const size_t start = idx * 64;
+          while (word) {
+            auto index = start + __builtin_ctzll(word);
+            bits::setBit(rawResult, index, cmp(rawA, rawB, index));
+            word &= word - 1;
+          }
+        },
+        [&](int32_t idx) {
+          auto word = rowsData[idx];
+          const size_t start = idx * 64;
+          if (kAllSet == word) {
+            // Do 64 comparisons in a batch, set results by SIMD.
+            for (size_t row = 0; row < 64; ++row) {
+              resultData[row] = cmp(rawA, rawB, row + start) ? -1 : 0;
+            }
+            rawResult[idx] = to64Bits(resultData);
+          } else {
+            while (word) {
+              auto index = __builtin_ctzll(word);
+              resultData[index] = cmp(rawA, rawB, start + index) ? -1 : 0;
+              word &= word - 1;
+            }
+            // Set results only for selected rows.
+            uint64_t mask = rowsData[idx];
+            rawResult[idx] =
+                (rawResult[idx] & ~mask) | (to64Bits(resultData) & mask);
+          }
+        });
+  }
+}
+} // namespace detail
+
+template <
+    typename T,
+    bool kIsLeftConstant,
+    bool kIsRightConstant,
+    typename ComparisonOp>
+void applySimdComparison(
+    const vector_size_t begin,
+    const vector_size_t end,
+    const T* rawLhs,
+    const T* rawRhs,
+    uint8_t* rawResult) {
+  using d_type = xsimd::batch<T>;
+  constexpr auto numScalarElements = d_type::size;
+  const auto vectorEnd = (end - begin) - (end - begin) % numScalarElements;
+  static_assert(
+      numScalarElements == 2 || numScalarElements == 4 ||
+          numScalarElements == 8 || numScalarElements == 16 ||
+          numScalarElements == 32,
+      "Unsupported number of scalar elements");
+  if constexpr (numScalarElements == 2 || numScalarElements == 4) {
+    for (auto i = begin; i < vectorEnd; i += 8) {
+      rawResult[i / 8] = 0;
+      for (auto j = 0; j < 8 && (i + j) < vectorEnd; j += numScalarElements) {
+        auto left = detail::loadSimdData<T, kIsLeftConstant>(rawLhs, i + j);
+        auto right = detail::loadSimdData<T, kIsRightConstant>(rawRhs, i + j);
+
+        uint8_t res = simd::toBitMask(ComparisonOp()(left, right));
+        rawResult[i / 8] |= res << j;
+      }
+    }
+  } else {
+    for (auto i = begin; i < vectorEnd; i += numScalarElements) {
+      auto left = detail::loadSimdData<T, kIsLeftConstant>(rawLhs, i);
+      auto right = detail::loadSimdData<T, kIsRightConstant>(rawRhs, i);
+
+      auto res = simd::toBitMask(ComparisonOp()(left, right));
+      if constexpr (numScalarElements == 8) {
+        rawResult[i / 8] = res;
+      } else if constexpr (numScalarElements == 16) {
+        uint16_t* addr = reinterpret_cast<uint16_t*>(rawResult + i / 8);
+        *addr = res;
+      } else if constexpr (numScalarElements == 32) {
+        uint32_t* addr = reinterpret_cast<uint32_t*>(rawResult + i / 8);
+        *addr = res;
+      }
+    }
+  }
+
+  // Evaluate remaining values.
+  for (auto i = vectorEnd; i < end; i++) {
+    if constexpr (kIsRightConstant && kIsLeftConstant) {
+      bits::setBit(rawResult, i, ComparisonOp()(rawLhs[0], rawRhs[0]));
+    } else if constexpr (kIsRightConstant) {
+      bits::setBit(rawResult, i, ComparisonOp()(rawLhs[i], rawRhs[0]));
+    } else if constexpr (kIsLeftConstant) {
+      bits::setBit(rawResult, i, ComparisonOp()(rawLhs[0], rawRhs[i]));
+    } else {
+      bits::setBit(rawResult, i, ComparisonOp()(rawLhs[i], rawRhs[i]));
+    }
+  }
+}
+
+template <typename T, typename ComparisonOp>
+void applySimdComparison(
+    const SelectivityVector& rows,
+    std::vector<VectorPtr>& args,
+    VectorPtr& result) {
+  auto resultVector = result->asUnchecked<FlatVector<bool>>();
+  auto rawResult = resultVector->mutableRawValues<uint8_t>();
+  if (args[0]->isConstantEncoding() && args[1]->isConstantEncoding()) {
+    auto l = args[0]->asUnchecked<ConstantVector<T>>()->valueAt(0);
+    auto r = args[1]->asUnchecked<ConstantVector<T>>()->valueAt(0);
+    applySimdComparison<T, true, true, ComparisonOp>(
+        rows.begin(), rows.end(), &l, &r, rawResult);
+  } else if (args[0]->isConstantEncoding()) {
+    auto l = args[0]->asUnchecked<ConstantVector<T>>()->valueAt(0);
+    auto rawRhs = args[1]->asUnchecked<FlatVector<T>>()->rawValues();
+    applySimdComparison<T, true, false, ComparisonOp>(
+        rows.begin(), rows.end(), &l, rawRhs, rawResult);
+  } else if (args[1]->isConstantEncoding()) {
+    auto rawLhs = args[0]->asUnchecked<FlatVector<T>>()->rawValues();
+    auto r = args[1]->asUnchecked<ConstantVector<T>>()->valueAt(0);
+    applySimdComparison<T, false, true, ComparisonOp>(
+        rows.begin(), rows.end(), rawLhs, &r, rawResult);
+  } else {
+    auto rawLhs = args[0]->asUnchecked<FlatVector<T>>()->rawValues();
+    auto rawRhs = args[1]->asUnchecked<FlatVector<T>>()->rawValues();
+    applySimdComparison<T, false, false, ComparisonOp>(
+        rows.begin(), rows.end(), rawLhs, rawRhs, rawResult);
+  }
+}
+
+template <typename A, typename B, typename Compare, typename... Args>
+void applyAutoSimdComparison(
+    const SelectivityVector& rows,
+    std::vector<VectorPtr>& args,
+    VectorPtr& result,
+    Args... cmpArgs) {
+  const Compare cmp;
+  if (args[0]->isFlatEncoding() && args[1]->isFlatEncoding()) {
+    const A* __restrict rawA =
+        args[0]->asUnchecked<FlatVector<A>>()->template rawValues<A>();
+    const B* __restrict rawB =
+        args[1]->asUnchecked<FlatVector<B>>()->template rawValues<B>();
+    detail::applyAutoSimdComparisonInternal(
+        rows,
+        rawA,
+        rawB,
+        [&](const A* __restrict rawA, const B* __restrict rawB, int i) {
+          if constexpr (sizeof...(cmpArgs) > 0) {
+            return Compare::apply(rawA[i], rawB[i], cmpArgs...);
+          } else {
+            return cmp(rawA[i], rawB[i]);
+          }
+        },
+        result);
+  } else if (args[0]->isConstantEncoding() && args[1]->isFlatEncoding()) {
+    const A constA = args[0]->asUnchecked<ConstantVector<A>>()->valueAt(0);
+    const A* __restrict rawA = &constA;
+    const B* __restrict rawB =
+        args[1]->asUnchecked<FlatVector<B>>()->template rawValues<B>();
+    detail::applyAutoSimdComparisonInternal(
+        rows,
+        rawA,
+        rawB,
+        [&](const A* __restrict rawA, const B* __restrict rawB, int i) {
+          if constexpr (sizeof...(cmpArgs) > 0) {
+            return Compare::apply(rawA[0], rawB[i], cmpArgs...);
+          } else {
+            return cmp(rawA[0], rawB[i]);
+          }
+        },
+        result);
+  } else if (args[0]->isFlatEncoding() && args[1]->isConstantEncoding()) {
+    const A* __restrict rawA =
+        args[0]->asUnchecked<FlatVector<A>>()->template rawValues<A>();
+    const B constB = args[1]->asUnchecked<ConstantVector<B>>()->valueAt(0);
+    const B* __restrict rawB = &constB;
+    detail::applyAutoSimdComparisonInternal(
+        rows,
+        rawA,
+        rawB,
+        [&](const A* __restrict rawA, const B* __restrict rawB, int i) {
+          if constexpr (sizeof...(cmpArgs) > 0) {
+            return Compare::apply(rawA[i], rawB[0], cmpArgs...);
+          } else {
+            return cmp(rawA[i], rawB[0]);
+          }
+        },
+        result);
+  } else if (args[0]->isConstantEncoding() && args[1]->isConstantEncoding()) {
+    const A constA = args[0]->asUnchecked<ConstantVector<A>>()->valueAt(0);
+    const A* __restrict rawA = &constA;
+    const B constB = args[1]->asUnchecked<ConstantVector<B>>()->valueAt(0);
+    const B* __restrict rawB = &constB;
+    detail::applyAutoSimdComparisonInternal(
+        rows,
+        rawA,
+        rawB,
+        [&](const A* __restrict rawA, const B* __restrict rawB, int i) {
+          if constexpr (sizeof...(cmpArgs) > 0) {
+            return Compare::apply(rawA[0], rawB[0], cmpArgs...);
+          } else {
+            return cmp(rawA[0], rawB[0]);
+          }
+        },
+        result);
+  } else {
+    VELOX_UNREACHABLE();
+  }
+}
+} // namespace facebook::velox::functions
diff --git a/velox/functions/prestosql/Comparisons.cpp b/velox/functions/prestosql/Comparisons.cpp
index 05eab2b80f56..ef5e76d773a5 100644
--- a/velox/functions/prestosql/Comparisons.cpp
+++ b/velox/functions/prestosql/Comparisons.cpp
@@ -17,6 +17,7 @@
 #include "velox/functions/prestosql/Comparisons.h"
 #include <velox/common/base/Exceptions.h>
 #include "velox/functions/Udf.h"
+#include "velox/functions/lib/SIMDComparisonUtil.h"
 #include "velox/vector/BaseVector.h"
 
 namespace facebook::velox::functions {
@@ -36,71 +37,6 @@ namespace {
 /// If the vector encoding is not flat, we revert to non simd approach.
 template <typename ComparisonOp, typename Arch = xsimd::default_arch>
 struct SimdComparator {
-  template <typename T, bool isConstant>
-  inline auto loadSimdData(const T* rawData, vector_size_t offset) {
-    using d_type = xsimd::batch<T>;
-    if constexpr (isConstant) {
-      return xsimd::broadcast<T>(rawData[0]);
-    }
-    return d_type::load_unaligned(rawData + offset);
-  }
-
-  template <typename T, bool isLeftConstant, bool isRightConstant>
-  void applySimdComparison(
-      const vector_size_t begin,
-      const vector_size_t end,
-      const T* rawLhs,
-      const T* rawRhs,
-      uint8_t* rawResult) {
-    using d_type = xsimd::batch<T>;
-    constexpr auto numScalarElements = d_type::size;
-    const auto vectorEnd = (end - begin) - (end - begin) % numScalarElements;
-
-    if constexpr (numScalarElements == 2 || numScalarElements == 4) {
-      for (auto i = begin; i < vectorEnd; i += 8) {
-        rawResult[i / 8] = 0;
-        for (auto j = 0; j < 8 && (i + j) < vectorEnd; j += numScalarElements) {
-          auto left = loadSimdData<T, isLeftConstant>(rawLhs, i + j);
-          auto right = loadSimdData<T, isRightConstant>(rawRhs, i + j);
-
-          uint8_t res = simd::toBitMask(ComparisonOp()(left, right));
-          rawResult[i / 8] |= res << j;
-        }
-      }
-    } else {
-      for (auto i = begin; i < vectorEnd; i += numScalarElements) {
-        auto left = loadSimdData<T, isLeftConstant>(rawLhs, i);
-        auto right = loadSimdData<T, isRightConstant>(rawRhs, i);
-
-        auto res = simd::toBitMask(ComparisonOp()(left, right));
-        if constexpr (numScalarElements == 8) {
-          rawResult[i / 8] = res;
-        } else if constexpr (numScalarElements == 16) {
-          uint16_t* addr = reinterpret_cast<uint16_t*>(rawResult + i / 8);
-          *addr = res;
-        } else if constexpr (numScalarElements == 32) {
-          uint32_t* addr = reinterpret_cast<uint32_t*>(rawResult + i / 8);
-          *addr = res;
-        } else {
-          VELOX_FAIL("Unsupported number of scalar elements");
-        }
-      }
-    }
-
-    // Evaluate remaining values.
-    for (auto i = vectorEnd; i < end; i++) {
-      if constexpr (isRightConstant && isLeftConstant) {
-        bits::setBit(rawResult, i, ComparisonOp()(rawLhs[0], rawRhs[0]));
-      } else if constexpr (isRightConstant) {
-        bits::setBit(rawResult, i, ComparisonOp()(rawLhs[i], rawRhs[0]));
-      } else if constexpr (isLeftConstant) {
-        bits::setBit(rawResult, i, ComparisonOp()(rawLhs[0], rawRhs[i]));
-      } else {
-        bits::setBit(rawResult, i, ComparisonOp()(rawLhs[i], rawRhs[i]));
-      }
-    }
-  }
-
   template <typename T>
   inline bool compare(T& l, T& r) const {
     if constexpr (std::is_floating_point_v<T>) {
@@ -167,22 +103,22 @@ struct SimdComparator {
       if (lhs.isConstantEncoding() && rhs.isConstantEncoding()) {
         auto l = lhs.asUnchecked<ConstantVector<T>>()->valueAt(0);
         auto r = rhs.asUnchecked<ConstantVector<T>>()->valueAt(0);
-        applySimdComparison<T, true, true>(
+        applySimdComparison<T, true, true, ComparisonOp>(
             rows.begin(), rows.end(), &l, &r, rawResult);
       } else if (lhs.isConstantEncoding()) {
         auto l = lhs.asUnchecked<ConstantVector<T>>()->valueAt(0);
         auto rawRhs = rhs.asUnchecked<FlatVector<T>>()->rawValues();
-        applySimdComparison<T, true, false>(
+        applySimdComparison<T, true, false, ComparisonOp>(
             rows.begin(), rows.end(), &l, rawRhs, rawResult);
       } else if (rhs.isConstantEncoding()) {
         auto rawLhs = lhs.asUnchecked<FlatVector<T>>()->rawValues();
         auto r = rhs.asUnchecked<ConstantVector<T>>()->valueAt(0);
-        applySimdComparison<T, false, true>(
+        applySimdComparison<T, false, true, ComparisonOp>(
             rows.begin(), rows.end(), rawLhs, &r, rawResult);
       } else {
         auto rawLhs = lhs.asUnchecked<FlatVector<T>>()->rawValues();
         auto rawRhs = rhs.asUnchecked<FlatVector<T>>()->rawValues();
-        applySimdComparison<T, false, false>(
+        applySimdComparison<T, false, false, ComparisonOp>(
             rows.begin(), rows.end(), rawLhs, rawRhs, rawResult);
       }
 
diff --git a/velox/functions/sparksql/Comparisons.cpp b/velox/functions/sparksql/Comparisons.cpp
index 5676503fe9f5..f52e603da816 100644
--- a/velox/functions/sparksql/Comparisons.cpp
+++ b/velox/functions/sparksql/Comparisons.cpp
@@ -17,6 +17,7 @@
 
 #include "velox/expression/EvalCtx.h"
 #include "velox/expression/Expr.h"
+#include "velox/functions/lib/SIMDComparisonUtil.h"
 #include "velox/functions/sparksql/Comparisons.h"
 #include "velox/type/Type.h"
 
@@ -37,28 +38,45 @@ class ComparisonFunction final : public exec::VectorFunction {
       const TypePtr& outputType,
       exec::EvalCtx& context,
       VectorPtr& result) const override {
+    const Cmp cmp;
     context.ensureWritable(rows, BOOLEAN(), result);
+    result->clearNulls(rows);
+    if ((args[0]->isFlatEncoding() || args[0]->isConstantEncoding()) &&
+        (args[1]->isFlatEncoding() || args[1]->isConstantEncoding())) {
+      if constexpr (
+          kind == TypeKind::TINYINT || kind == TypeKind::SMALLINT ||
+          kind == TypeKind::INTEGER || kind == TypeKind::BIGINT) {
+        if (rows.isAllSelected()) {
+          applySimdComparison<T, Cmp>(rows, args, result);
+          return;
+        }
+      }
+      if (rows.end() - rows.begin() > 64) {
+        applyAutoSimdComparison<T, T, Cmp>(rows, args, result);
+        return;
+      }
+    }
+
     auto* flatResult = result->asUnchecked<FlatVector<bool>>();
-    const Cmp cmp;
 
     if (args[0]->isFlatEncoding() && args[1]->isFlatEncoding()) {
       // Fast path for (flat, flat).
-      auto rawA = args[0]->asUnchecked<FlatVector<T>>()->mutableRawValues();
-      auto rawB = args[1]->asUnchecked<FlatVector<T>>()->mutableRawValues();
+      const auto* rawA = args[0]->asUnchecked<FlatVector<T>>()->rawValues();
+      const auto* rawB = args[1]->asUnchecked<FlatVector<T>>()->rawValues();
       rows.applyToSelected(
           [&](vector_size_t i) { flatResult->set(i, cmp(rawA[i], rawB[i])); });
     } else if (args[0]->isConstantEncoding() && args[1]->isFlatEncoding()) {
       // Fast path for (const, flat).
       auto constant = args[0]->asUnchecked<ConstantVector<T>>()->valueAt(0);
-      auto rawValues =
-          args[1]->asUnchecked<FlatVector<T>>()->mutableRawValues();
+      const auto* rawValues =
+          args[1]->asUnchecked<FlatVector<T>>()->rawValues();
       rows.applyToSelected([&](vector_size_t i) {
         flatResult->set(i, cmp(constant, rawValues[i]));
       });
     } else if (args[0]->isFlatEncoding() && args[1]->isConstantEncoding()) {
       // Fast path for (flat, const).
-      auto rawValues =
-          args[0]->asUnchecked<FlatVector<T>>()->mutableRawValues();
+      const auto* rawValues =
+          args[0]->asUnchecked<FlatVector<T>>()->rawValues();
       auto constant = args[1]->asUnchecked<ConstantVector<T>>()->valueAt(0);
       rows.applyToSelected([&](vector_size_t i) {
         flatResult->set(i, cmp(rawValues[i], constant));
@@ -136,7 +154,7 @@ class BoolComparisonFunction final : public exec::VectorFunction {
   }
 };
 
-template <template <typename> class Cmp>
+template <template <typename> class Cmp, typename StdCmp>
 std::shared_ptr<exec::VectorFunction> makeImpl(
     const std::string& functionName,
     const std::vector<exec::VectorFunctionArg>& args) {
@@ -150,17 +168,21 @@ std::shared_ptr<exec::VectorFunction> makeImpl(
     return std::make_shared<ComparisonFunction<      \
         Cmp<TypeTraits<TypeKind::kind>::NativeType>, \
         TypeKind::kind>>();
-    SCALAR_CASE(TINYINT)
-    SCALAR_CASE(SMALLINT)
-    SCALAR_CASE(INTEGER)
-    SCALAR_CASE(BIGINT)
-    SCALAR_CASE(HUGEINT)
     SCALAR_CASE(REAL)
     SCALAR_CASE(DOUBLE)
+    SCALAR_CASE(HUGEINT)
     SCALAR_CASE(VARCHAR)
     SCALAR_CASE(VARBINARY)
     SCALAR_CASE(TIMESTAMP)
 #undef SCALAR_CASE
+#define STD_SCALAR_CASE(kind) \
+  case TypeKind::kind:        \
+    return std::make_shared<ComparisonFunction<StdCmp, TypeKind::kind>>();
+    STD_SCALAR_CASE(TINYINT)
+    STD_SCALAR_CASE(SMALLINT)
+    STD_SCALAR_CASE(INTEGER)
+    STD_SCALAR_CASE(BIGINT)
+#undef STDSCALAR_CASE
     case TypeKind::BOOLEAN:
       return std::make_shared<BoolComparisonFunction<
           Cmp<TypeTraits<TypeKind::BOOLEAN>::NativeType>>>();
@@ -240,35 +262,35 @@ std::shared_ptr<exec::VectorFunction> makeEqualTo(
     const std::string& functionName,
     const std::vector<exec::VectorFunctionArg>& args,
     const core::QueryConfig& /*config*/) {
-  return makeImpl<Equal>(functionName, args);
+  return makeImpl<Equal, std::equal_to<>>(functionName, args);
 }
 
 std::shared_ptr<exec::VectorFunction> makeLessThan(
     const std::string& functionName,
     const std::vector<exec::VectorFunctionArg>& args,
     const core::QueryConfig& /*config*/) {
-  return makeImpl<Less>(functionName, args);
+  return makeImpl<Less, std::less<>>(functionName, args);
 }
 
 std::shared_ptr<exec::VectorFunction> makeGreaterThan(
     const std::string& functionName,
     const std::vector<exec::VectorFunctionArg>& args,
     const core::QueryConfig& /*config*/) {
-  return makeImpl<Greater>(functionName, args);
+  return makeImpl<Greater, std::greater<>>(functionName, args);
 }
 
 std::shared_ptr<exec::VectorFunction> makeLessThanOrEqual(
     const std::string& functionName,
     const std::vector<exec::VectorFunctionArg>& args,
     const core::QueryConfig& /*config*/) {
-  return makeImpl<LessOrEqual>(functionName, args);
+  return makeImpl<LessOrEqual, std::less_equal<>>(functionName, args);
 }
 
 std::shared_ptr<exec::VectorFunction> makeGreaterThanOrEqual(
     const std::string& functionName,
     const std::vector<exec::VectorFunctionArg>& args,
     const core::QueryConfig& /*config*/) {
-  return makeImpl<GreaterOrEqual>(functionName, args);
+  return makeImpl<GreaterOrEqual, std::greater_equal<>>(functionName, args);
 }
 
 std::shared_ptr<exec::VectorFunction> makeEqualToNullSafe(
diff --git a/velox/functions/sparksql/DecimalCompare.cpp b/velox/functions/sparksql/DecimalCompare.cpp
index c4ddbb0b5815..940e1d486ea8 100644
--- a/velox/functions/sparksql/DecimalCompare.cpp
+++ b/velox/functions/sparksql/DecimalCompare.cpp
@@ -16,6 +16,7 @@
 
 #include "velox/expression/DecodedArgs.h"
 #include "velox/expression/VectorFunction.h"
+#include "velox/functions/lib/SIMDComparisonUtil.h"
 #include "velox/functions/sparksql/DecimalUtil.h"
 
 namespace facebook::velox::functions::sparksql {
@@ -115,6 +116,13 @@ class DecimalCompareFunction : public exec::VectorFunction {
       exec::EvalCtx& context,
       VectorPtr& result) const override {
     prepareResults(rows, resultType, context, result);
+    if ((args[0]->isFlatEncoding() || args[0]->isConstantEncoding()) &&
+        (args[1]->isFlatEncoding() || args[1]->isConstantEncoding()) &&
+        rows.end() - rows.begin() > 64) {
+      applyAutoSimdComparison<A, B, Operation, int8_t, bool>(
+          rows, args, result, deltaScale_, need256_);
+      return;
+    }
 
     // Fast path when the first argument is a flat vector.
     if (args[0]->isFlatEncoding()) {
diff --git a/velox/functions/sparksql/benchmarks/CMakeLists.txt b/velox/functions/sparksql/benchmarks/CMakeLists.txt
index 837c983b7df8..d191483297ab 100644
--- a/velox/functions/sparksql/benchmarks/CMakeLists.txt
+++ b/velox/functions/sparksql/benchmarks/CMakeLists.txt
@@ -28,6 +28,11 @@ add_executable(velox_sparksql_benchmarks_compare CompareBenchmark.cpp)
 target_link_libraries(velox_sparksql_benchmarks_compare velox_functions_spark
                       velox_benchmark_builder velox_vector_test_lib)
 
+add_executable(velox_sparksql_benchmarks_simd_compare SIMDCompareBenchmark.cpp)
+target_link_libraries(
+  velox_sparksql_benchmarks_simd_compare velox_functions_spark
+  velox_benchmark_builder velox_vector_test_lib)
+
 add_executable(velox_sparksql_benchmarks_hash HashBenchmark.cpp)
 target_link_libraries(velox_sparksql_benchmarks_hash velox_functions_spark
                       velox_benchmark_builder velox_vector_test_lib)
diff --git a/velox/functions/sparksql/benchmarks/CompareBenchmark.cpp b/velox/functions/sparksql/benchmarks/CompareBenchmark.cpp
index e4d3c9f3f608..b0b7efdfc3cb 100644
--- a/velox/functions/sparksql/benchmarks/CompareBenchmark.cpp
+++ b/velox/functions/sparksql/benchmarks/CompareBenchmark.cpp
@@ -30,30 +30,34 @@ int main(int argc, char** argv) {
   functions::sparksql::registerFunctions("");
 
   ExpressionBenchmarkBuilder benchmarkBuilder;
-  benchmarkBuilder
-      .addBenchmarkSet(
-          "compare", ROW({"c0", "c1"}, {DECIMAL(18, 6), DECIMAL(38, 16)}))
-      .withFuzzerOptions({.vectorSize = 1000, .nullRatio = 0.1})
-      .addExpression("gt", "decimal_greaterthan(c0, c1)")
-      .addExpression(
-          "gt_with_cast", "greaterthan(cast (c0 as decimal(38, 16)), c1)")
-      .addExpression("gte", "decimal_greaterthanorequal(c0, c1)")
-      .addExpression(
-          "gte_with_cast",
-          "greaterthanorequal(cast (c0 as decimal(38, 16)), c1)")
-      .addExpression("lt", "decimal_lessthan(c0, c1)")
-      .addExpression(
-          "lt_with_cast", "lessthan(cast (c0 as decimal(38, 16)), c1)")
-      .addExpression("lte", "decimal_lessthanorequal(c0, c1)")
-      .addExpression(
-          "lte_with_cast", "lessthanorequal(cast (c0 as decimal(38, 16)), c1)")
-      .addExpression("eq", "decimal_equalto(c0, c1)")
-      .addExpression(
-          "eq_with_cast", "equalto(cast (c0 as decimal(38, 16)), c1)")
-      .addExpression("neq", "decimal_notequalto(c0, c1)")
-      .addExpression(
-          "neq_with_cast", "not(equalto(cast (c0 as decimal(38, 16)), c1))")
-      .withIterations(100);
+  for (auto nullRatio : {0.0, 0.1, 0.9}) {
+    benchmarkBuilder
+        .addBenchmarkSet(
+            fmt::format("compare#{}\%null", nullRatio * 100),
+            ROW({"c0", "c1"}, {DECIMAL(18, 6), DECIMAL(38, 16)}))
+        .withFuzzerOptions({.vectorSize = 1000, .nullRatio = nullRatio})
+        .addExpression("gt", "decimal_greaterthan(c0, c1)")
+        .addExpression(
+            "gt_with_cast", "greaterthan(cast (c0 as decimal(38, 16)), c1)")
+        .addExpression("gte", "decimal_greaterthanorequal(c0, c1)")
+        .addExpression(
+            "gte_with_cast",
+            "greaterthanorequal(cast (c0 as decimal(38, 16)), c1)")
+        .addExpression("lt", "decimal_lessthan(c0, c1)")
+        .addExpression(
+            "lt_with_cast", "lessthan(cast (c0 as decimal(38, 16)), c1)")
+        .addExpression("lte", "decimal_lessthanorequal(c0, c1)")
+        .addExpression(
+            "lte_with_cast",
+            "lessthanorequal(cast (c0 as decimal(38, 16)), c1)")
+        .addExpression("eq", "decimal_equalto(c0, c1)")
+        .addExpression(
+            "eq_with_cast", "equalto(cast (c0 as decimal(38, 16)), c1)")
+        .addExpression("neq", "decimal_notequalto(c0, c1)")
+        .addExpression(
+            "neq_with_cast", "not(equalto(cast (c0 as decimal(38, 16)), c1))")
+        .withIterations(100);
+  }
 
   benchmarkBuilder.registerBenchmarks();
   folly::runBenchmarks();
diff --git a/velox/functions/sparksql/benchmarks/SIMDCompareBenchmark.cpp b/velox/functions/sparksql/benchmarks/SIMDCompareBenchmark.cpp
new file mode 100644
index 000000000000..31a234f6dc4c
--- /dev/null
+++ b/velox/functions/sparksql/benchmarks/SIMDCompareBenchmark.cpp
@@ -0,0 +1,60 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License 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.
+ */
+
+#include <folly/Benchmark.h>
+#include <folly/init/Init.h>
+
+#include "velox/benchmarks/ExpressionBenchmarkBuilder.h"
+#include "velox/functions/sparksql/Register.h"
+
+using namespace facebook;
+
+using namespace facebook::velox;
+
+int main(int argc, char** argv) {
+  folly::Init init(&argc, &argv);
+  memory::MemoryManager::initialize({});
+  functions::sparksql::registerFunctions("");
+  std::vector<TypePtr> inputTypes = {
+      TINYINT(),
+      SMALLINT(),
+      INTEGER(),
+      BIGINT(),
+      REAL(),
+      DOUBLE(),
+      TIMESTAMP(),
+      VARCHAR(),
+  };
+  ExpressionBenchmarkBuilder benchmarkBuilder;
+  for (auto nullRatio : {0.0, 0.1, 0.9}) {
+    for (auto& inputType : inputTypes) {
+      benchmarkBuilder
+          .addBenchmarkSet(
+              fmt::format(
+                  "{}#{}\%null", inputType->toString(), nullRatio * 100),
+              ROW({"c0", "c1"}, {inputType, inputType}))
+          .withFuzzerOptions({.vectorSize = 10000, .nullRatio = nullRatio})
+          .addExpression("equalto", "equalto(c0, c1)")
+          .addExpression("greaterthan", "greaterthan(c0, c1)")
+          .addExpression("greaterthanorequal", "greaterthanorequal(c0, c1)")
+          .withIterations(100);
+    }
+  }
+
+  benchmarkBuilder.registerBenchmarks();
+  folly::runBenchmarks();
+  return 0;
+}
diff --git a/velox/functions/sparksql/tests/ComparisonsTest.cpp b/velox/functions/sparksql/tests/ComparisonsTest.cpp
index aab54943084d..1c1c181a7693 100644
--- a/velox/functions/sparksql/tests/ComparisonsTest.cpp
+++ b/velox/functions/sparksql/tests/ComparisonsTest.cpp
@@ -66,6 +66,59 @@ class ComparisonsTest : public SparkFunctionBaseTest {
     return evaluateOnce<bool>("greaterthanorequal(c0, c1)", a, b);
   }
 
+  template <TypeKind kind>
+  void runSIMDCompareAndAssert(int size, int unSelectedRows = 0) {
+    using T = typename TypeTraits<kind>::NativeType;
+    auto type = TypeTraits<kind>::ImplType::create();
+    auto expectedResult = std::vector<bool>();
+    expectedResult.resize(size);
+    VectorFuzzer::Options opts;
+    opts.vectorSize = size;
+    VectorFuzzer fuzzer(opts, execCtx_.pool());
+    auto lVector = fuzzer.fuzzFlat(type, size);
+    auto left = lVector->template as<FlatVector<T>>()->rawValues();
+    auto rVector = fuzzer.fuzzFlat(type, size);
+    auto right = rVector->template as<FlatVector<T>>()->rawValues();
+    auto constVector = fuzzer.fuzzConstant(type);
+    auto constant = constVector->template as<ConstantVector<T>>()->value();
+    SelectivityVector rows(size);
+    rows.setValidRange(0, unSelectedRows, false);
+
+    // Flat, Flat
+    std::vector<VectorPtr> childrenVectors = {lVector, rVector};
+    auto rowVector =
+        fuzzer.fuzzRow(std::move(childrenVectors), {"c0", "c1"}, size);
+    auto result =
+        evaluate<SimpleVector<bool>>("greaterthanorequal(c0, c1)", rowVector);
+    for (auto i = unSelectedRows; i < size; i++) {
+      expectedResult[i] = left[i] >= right[i];
+    }
+    velox::test::assertEqualVectors(
+        makeFlatVector<bool>(expectedResult), result, rows);
+
+    // Flat, Const
+    std::vector<VectorPtr> rConstVectors = {lVector, constVector};
+    rowVector = fuzzer.fuzzRow(std::move(rConstVectors), {"c0", "c1"}, size);
+    result =
+        evaluate<SimpleVector<bool>>("greaterthanorequal(c0, c1)", rowVector);
+    for (auto i = unSelectedRows; i < size; i++) {
+      expectedResult[i] = left[i] >= constant;
+    }
+    velox::test::assertEqualVectors(
+        makeFlatVector<bool>(expectedResult), result, rows);
+
+    // Const, Flat
+    std::vector<VectorPtr> lConstVectors = {constVector, rVector};
+    rowVector = fuzzer.fuzzRow(std::move(lConstVectors), {"c0", "c1"}, size);
+    result =
+        evaluate<SimpleVector<bool>>("greaterthanorequal(c0, c1)", rowVector);
+    for (auto i = unSelectedRows; i < size; i++) {
+      expectedResult[i] = constant >= right[i];
+    }
+    velox::test::assertEqualVectors(
+        makeFlatVector<bool>(expectedResult), result, rows);
+  }
+
   void runAndCompare(
       const std::string& functionName,
       const std::vector<VectorPtr>& input,
@@ -388,6 +441,32 @@ TEST_F(ComparisonsTest, testflat) {
   facebook::velox::test::assertEqualVectors(vectorBool0, actualBoolResult);
 }
 
+TEST_F(ComparisonsTest, testSIMDComparsion) {
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(64);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(65);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(1001);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(1001, 27);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(1001, 47);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(1001, 56);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(1001, 100);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(4096);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(4096, 30);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(4096, 31);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(4096, 32);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(4096, 33);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(4096, 78);
+
+  runSIMDCompareAndAssert<TypeKind::TINYINT>(675);
+  runSIMDCompareAndAssert<TypeKind::SMALLINT>(10240);
+  runSIMDCompareAndAssert<TypeKind::INTEGER>(7799);
+  runSIMDCompareAndAssert<TypeKind::BIGINT>(8876);
+  runSIMDCompareAndAssert<TypeKind::REAL>(1000);
+  runSIMDCompareAndAssert<TypeKind::DOUBLE>(10000);
+  runSIMDCompareAndAssert<TypeKind::TIMESTAMP>(1024);
+  runSIMDCompareAndAssert<TypeKind::VARCHAR>(686);
+  runSIMDCompareAndAssert<TypeKind::VARBINARY>(777);
+}
+
 TEST_F(ComparisonsTest, lessthan) {
   EXPECT_EQ(lessthan<int64_t>(1, 1), false);
   EXPECT_EQ(lessthan<int32_t>(1, 2), true);
diff --git a/velox/functions/sparksql/tests/DecimalCompareTest.cpp b/velox/functions/sparksql/tests/DecimalCompareTest.cpp
index 7201fd56508e..df6911b20948 100644
--- a/velox/functions/sparksql/tests/DecimalCompareTest.cpp
+++ b/velox/functions/sparksql/tests/DecimalCompareTest.cpp
@@ -24,9 +24,14 @@ class DecimalCompareTest : public SparkFunctionBaseTest {
   void testCompareExpr(
       const std::string& exprStr,
       const std::vector<VectorPtr>& input,
-      const VectorPtr& expectedResult) {
-    auto actual = evaluate(exprStr, makeRowVector(input));
-    velox::test::assertEqualVectors(expectedResult, actual);
+      const VectorPtr& expectedResult,
+      const std::optional<SelectivityVector>& rows = std::nullopt) {
+    auto actual = evaluate(exprStr, makeRowVector(input), rows);
+    if (rows.has_value()) {
+      velox::test::assertEqualVectors(expectedResult, actual, rows.value());
+    } else {
+      velox::test::assertEqualVectors(expectedResult, actual);
+    }
   }
 };
 
@@ -126,6 +131,45 @@ TEST_F(DecimalCompareTest, gt) {
               {100000, std::nullopt, 130000, 350000}, DECIMAL(6, 1)),
       },
       makeNullableFlatVector<bool>({false, std::nullopt, false, false}));
+
+  // All rows selected.
+  testCompareExpr(
+      "decimal_greaterthan(c0, c1)",
+      {
+          makeFlatVector<int64_t>(
+              70,
+              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
+              nullptr,
+              DECIMAL(6, 2)),
+          makeFlatVector<int64_t>(
+              70,
+              [](auto row) { return row % 2 == 0 ? 100 : 130; },
+              nullptr,
+              DECIMAL(5, 1)),
+      },
+      makeFlatVector<bool>(
+          70, [](auto row) { return row % 2 == 0 ? false : true; }));
+
+  // 90% rows selected.
+  SelectivityVector row(130);
+  row.setValidRange(0, 13, false);
+  testCompareExpr(
+      "decimal_greaterthan(c0, c1)",
+      {
+          makeFlatVector<int64_t>(
+              130,
+              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
+              nullptr,
+              DECIMAL(6, 2)),
+          makeFlatVector<int64_t>(
+              130,
+              [](auto row) { return row % 2 == 0 ? 100 : 130; },
+              nullptr,
+              DECIMAL(5, 1)),
+      },
+      makeFlatVector<bool>(
+          130, [](auto row) { return row % 2 == 0 ? false : true; }),
+      row);
 }
 
 TEST_F(DecimalCompareTest, gte) {
@@ -224,6 +268,43 @@ TEST_F(DecimalCompareTest, gte) {
               {100000, std::nullopt, 130000, 350000}, DECIMAL(6, 1)),
       },
       makeNullableFlatVector<bool>({false, std::nullopt, false, false}));
+
+  // All rows selected.
+  testCompareExpr(
+      "decimal_greaterthanorequal(c0, c1)",
+      {
+          makeFlatVector<int64_t>(
+              70,
+              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
+              nullptr,
+              DECIMAL(6, 2)),
+          makeFlatVector<int64_t>(
+              70,
+              [](auto row) { return row % 2 == 0 ? 100 : 130; },
+              nullptr,
+              DECIMAL(5, 1)),
+      },
+      makeFlatVector<bool>(70, [](auto /*row*/) { return true; }));
+
+  // 90% rows selected.
+  SelectivityVector row(130);
+  row.setValidRange(0, 13, false);
+  testCompareExpr(
+      "decimal_greaterthanorequal(c0, c1)",
+      {
+          makeFlatVector<int64_t>(
+              130,
+              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
+              nullptr,
+              DECIMAL(6, 2)),
+          makeFlatVector<int64_t>(
+              130,
+              [](auto row) { return row % 2 == 0 ? 100 : 130; },
+              nullptr,
+              DECIMAL(5, 1)),
+      },
+      makeFlatVector<bool>(130, [](auto /*row*/) { return true; }),
+      row);
 }
 
 TEST_F(DecimalCompareTest, eq) {
@@ -322,6 +403,45 @@ TEST_F(DecimalCompareTest, eq) {
               {100000, std::nullopt, 130000, 350000}, DECIMAL(6, 1)),
       },
       makeNullableFlatVector<bool>({false, std::nullopt, false, false}));
+
+  // All rows selected.
+  testCompareExpr(
+      "decimal_equalto(c0, c1)",
+      {
+          makeFlatVector<int64_t>(
+              70,
+              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
+              nullptr,
+              DECIMAL(6, 2)),
+          makeFlatVector<int64_t>(
+              70,
+              [](auto row) { return row % 2 == 0 ? 100 : 130; },
+              nullptr,
+              DECIMAL(5, 1)),
+      },
+      makeFlatVector<bool>(
+          70, [](auto row) { return row % 2 == 0 ? true : false; }));
+
+  // 90% rows selected.
+  SelectivityVector row(130);
+  row.setValidRange(0, 13, false);
+  testCompareExpr(
+      "decimal_equalto(c0, c1)",
+      {
+          makeFlatVector<int64_t>(
+              130,
+              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
+              nullptr,
+              DECIMAL(6, 2)),
+          makeFlatVector<int64_t>(
+              130,
+              [](auto row) { return row % 2 == 0 ? 100 : 130; },
+              nullptr,
+              DECIMAL(5, 1)),
+      },
+      makeFlatVector<bool>(
+          130, [](auto row) { return row % 2 == 0 ? true : false; }),
+      row);
 }
 
 TEST_F(DecimalCompareTest, neq) {
@@ -420,6 +540,45 @@ TEST_F(DecimalCompareTest, neq) {
               {100000, std::nullopt, 130000, 350000}, DECIMAL(6, 1)),
       },
       makeNullableFlatVector<bool>({true, std::nullopt, true, true}));
+
+  // All rows selected.
+  testCompareExpr(
+      "decimal_notequalto(c0, c1)",
+      {
+          makeFlatVector<int64_t>(
+              70,
+              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
+              nullptr,
+              DECIMAL(6, 2)),
+          makeFlatVector<int64_t>(
+              70,
+              [](auto row) { return row % 2 == 0 ? 100 : 130; },
+              nullptr,
+              DECIMAL(5, 1)),
+      },
+      makeFlatVector<bool>(
+          70, [](auto row) { return row % 2 == 0 ? false : true; }));
+
+  // 90% rows selected.
+  SelectivityVector row(130);
+  row.setValidRange(0, 13, false);
+  testCompareExpr(
+      "decimal_notequalto(c0, c1)",
+      {
+          makeFlatVector<int64_t>(
+              130,
+              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
+              nullptr,
+              DECIMAL(6, 2)),
+          makeFlatVector<int64_t>(
+              130,
+              [](auto row) { return row % 2 == 0 ? 100 : 130; },
+              nullptr,
+              DECIMAL(5, 1)),
+      },
+      makeFlatVector<bool>(
+          130, [](auto row) { return row % 2 == 0 ? false : true; }),
+      row);
 }
 
 TEST_F(DecimalCompareTest, lt) {
@@ -518,6 +677,43 @@ TEST_F(DecimalCompareTest, lt) {
               {100000, std::nullopt, 130000, 350000}, DECIMAL(6, 1)),
       },
       makeNullableFlatVector<bool>({true, std::nullopt, true, true}));
+
+  // All rows selected.
+  testCompareExpr(
+      "decimal_lessthan(c0, c1)",
+      {
+          makeFlatVector<int64_t>(
+              70,
+              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
+              nullptr,
+              DECIMAL(6, 2)),
+          makeFlatVector<int64_t>(
+              70,
+              [](auto row) { return row % 2 == 0 ? 100 : 130; },
+              nullptr,
+              DECIMAL(5, 1)),
+      },
+      makeFlatVector<bool>(70, [](auto /*row*/) { return false; }));
+
+  // 90% rows selected.
+  SelectivityVector row(130);
+  row.setValidRange(0, 13, false);
+  testCompareExpr(
+      "decimal_lessthan(c0, c1)",
+      {
+          makeFlatVector<int64_t>(
+              130,
+              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
+              nullptr,
+              DECIMAL(6, 2)),
+          makeFlatVector<int64_t>(
+              130,
+              [](auto row) { return row % 2 == 0 ? 100 : 130; },
+              nullptr,
+              DECIMAL(5, 1)),
+      },
+      makeFlatVector<bool>(130, [](auto /*row*/) { return false; }),
+      row);
 }
 
 TEST_F(DecimalCompareTest, lte) {
@@ -616,6 +812,45 @@ TEST_F(DecimalCompareTest, lte) {
               {100000, std::nullopt, 130000, 350000}, DECIMAL(6, 1)),
       },
       makeNullableFlatVector<bool>({true, std::nullopt, true, true}));
+
+  // All rows selected.
+  testCompareExpr(
+      "decimal_lessthanorequal(c0, c1)",
+      {
+          makeFlatVector<int64_t>(
+              70,
+              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
+              nullptr,
+              DECIMAL(6, 2)),
+          makeFlatVector<int64_t>(
+              70,
+              [](auto row) { return row % 2 == 0 ? 100 : 130; },
+              nullptr,
+              DECIMAL(5, 1)),
+      },
+      makeFlatVector<bool>(
+          70, [](auto row) { return row % 2 == 0 ? true : false; }));
+
+  // 90% rows selected.
+  SelectivityVector row(130);
+  row.setValidRange(0, 13, false);
+  testCompareExpr(
+      "decimal_lessthanorequal(c0, c1)",
+      {
+          makeFlatVector<int64_t>(
+              130,
+              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
+              nullptr,
+              DECIMAL(6, 2)),
+          makeFlatVector<int64_t>(
+              130,
+              [](auto row) { return row % 2 == 0 ? 100 : 130; },
+              nullptr,
+              DECIMAL(5, 1)),
+      },
+      makeFlatVector<bool>(
+          130, [](auto row) { return row % 2 == 0 ? true : false; }),
+      row);
 }
 
 } // namespace