From 1c31582714d978bc3d14e1082732bb45e3c55f2d Mon Sep 17 00:00:00 2001
From: mayan <yan.ma@intel.com>
Date: Mon, 10 Jul 2023 09:57:39 +0000
Subject: [PATCH] Implement Compare functions for Spark

---
 velox/functions/sparksql/CMakeLists.txt       |   1 +
 velox/functions/sparksql/Comparisons.cpp      | 155 +++++++++
 velox/functions/sparksql/Comparisons.h        |  84 +++++
 velox/functions/sparksql/RegisterCompare.cpp  |  30 +-
 velox/functions/sparksql/tests/CMakeLists.txt |   1 +
 .../functions/sparksql/tests/CompareTests.cpp | 298 ++++++++++++++++++
 6 files changed, 561 insertions(+), 8 deletions(-)
 create mode 100644 velox/functions/sparksql/Comparisons.cpp
 create mode 100644 velox/functions/sparksql/tests/CompareTests.cpp

diff --git a/velox/functions/sparksql/CMakeLists.txt b/velox/functions/sparksql/CMakeLists.txt
index b9ec0498a589..76ffce938e92 100644
--- a/velox/functions/sparksql/CMakeLists.txt
+++ b/velox/functions/sparksql/CMakeLists.txt
@@ -17,6 +17,7 @@ add_library(
   ArraySort.cpp
   Bitwise.cpp
   CompareFunctionsNullSafe.cpp
+  Comparisons.cpp
   Hash.cpp
   In.cpp
   LeastGreatest.cpp
diff --git a/velox/functions/sparksql/Comparisons.cpp b/velox/functions/sparksql/Comparisons.cpp
new file mode 100644
index 000000000000..4f1616917b35
--- /dev/null
+++ b/velox/functions/sparksql/Comparisons.cpp
@@ -0,0 +1,155 @@
+/*
+ * 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 "velox/functions/sparksql/LeastGreatest.h"
+
+#include "velox/expression/EvalCtx.h"
+#include "velox/expression/Expr.h"
+#include "velox/functions/sparksql/Comparisons.h"
+#include "velox/type/Type.h"
+
+namespace facebook::velox::functions::sparksql {
+namespace {
+
+template <typename Cmp, TypeKind kind>
+class ComparisonFunction final : public exec::VectorFunction {
+  using T = typename TypeTraits<kind>::NativeType;
+
+  bool isDefaultNullBehavior() const override {
+    return true;
+  }
+
+  bool supportsFlatNoNullsFastPath() const override {
+    return true;
+  }
+
+  void apply(
+      const SelectivityVector& rows,
+      std::vector<VectorPtr>& args,
+      const TypePtr& outputType,
+      exec::EvalCtx& context,
+      VectorPtr& result) const override {
+    context.ensureWritable(rows, BOOLEAN(), result);
+    auto* flatResult = result->asFlatVector<bool>();
+    flatResult->mutableRawValues<uint64_t>();
+    const Cmp cmp;
+    if (args[0]->isFlatEncoding() && args[1]->isFlatEncoding()) {
+      // Fast path for (flat, flat).
+      auto flatA = args[0]->asUnchecked<FlatVector<T>>();
+      auto rawA = flatA->mutableRawValues();
+      auto flatB = args[1]->asUnchecked<FlatVector<T>>();
+      auto rawB = flatB->mutableRawValues();
+      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<SimpleVector<T>>()->valueAt(0);
+      auto flatValues = args[1]->asUnchecked<FlatVector<T>>();
+      auto rawValues = flatValues->mutableRawValues();
+      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 flatValues = args[0]->asUnchecked<FlatVector<T>>();
+      auto constant = args[1]->asUnchecked<SimpleVector<T>>()->valueAt(0);
+      auto rawValues = flatValues->mutableRawValues();
+      rows.applyToSelected([&](vector_size_t i) {
+        flatResult->set(i, cmp(rawValues[i], constant));
+      });
+    } else {
+      // Fast path if one or more arguments are encoded.
+      exec::DecodedArgs decodedArgs(rows, args, context);
+      auto decoded0 = decodedArgs.at(0);
+      auto decoded1 = decodedArgs.at(1);
+      rows.applyToSelected([&](vector_size_t i) {
+        flatResult->set(
+            i, cmp(decoded0->valueAt<T>(i), decoded1->valueAt<T>(i)));
+      });
+    }
+  }
+};
+
+template <template <typename> class Cmp>
+std::shared_ptr<exec::VectorFunction> makeImpl(
+    const std::string& functionName,
+    const std::vector<exec::VectorFunctionArg>& args) {
+  VELOX_CHECK_EQ(args.size(), 2);
+  for (size_t i = 1; i < args.size(); i++) {
+    VELOX_CHECK(*args[i].type == *args[0].type);
+  }
+  switch (args[0].type->kind()) {
+#define SCALAR_CASE(kind)                            \
+  case TypeKind::kind:                               \
+    return std::make_shared<ComparisonFunction<      \
+        Cmp<TypeTraits<TypeKind::kind>::NativeType>, \
+        TypeKind::kind>>();
+    SCALAR_CASE(BOOLEAN)
+    SCALAR_CASE(TINYINT)
+    SCALAR_CASE(SMALLINT)
+    SCALAR_CASE(INTEGER)
+    SCALAR_CASE(BIGINT)
+    SCALAR_CASE(HUGEINT)
+    SCALAR_CASE(REAL)
+    SCALAR_CASE(DOUBLE)
+    SCALAR_CASE(VARCHAR)
+    SCALAR_CASE(VARBINARY)
+    SCALAR_CASE(TIMESTAMP)
+#undef SCALAR_CASE
+    default:
+      VELOX_NYI(
+          "{} does not support arguments of type {}",
+          functionName,
+          args[0].type->kind());
+  }
+}
+
+} // namespace
+
+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);
+}
+
+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);
+}
+
+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);
+}
+
+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);
+}
+
+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);
+}
+} // namespace facebook::velox::functions::sparksql
diff --git a/velox/functions/sparksql/Comparisons.h b/velox/functions/sparksql/Comparisons.h
index 1190fdce403c..e4e0caf5d922 100644
--- a/velox/functions/sparksql/Comparisons.h
+++ b/velox/functions/sparksql/Comparisons.h
@@ -13,6 +13,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+#pragma once
+
+#include <memory>
+
 #include "velox/expression/VectorFunction.h"
 
 namespace facebook::velox::functions::sparksql {
@@ -59,4 +63,84 @@ struct Equal {
   }
 };
 
+template <typename T>
+struct LessOrEqual {
+  constexpr bool operator()(const T& a, const T& b) const {
+    Less<T> less;
+    Equal<T> equal;
+    return less(a, b) || equal(a, b);
+  }
+};
+
+template <typename T>
+struct GreaterOrEqual : private Less<T> {
+  constexpr bool operator()(const T& a, const T& b) const {
+    Less<T> less;
+    Equal<T> equal;
+    return less(b, a) || equal(a, b);
+  }
+};
+
+/// Supported Types:
+/// TINYINT
+/// SMALLINT
+/// INTEGER
+/// BIGINT
+/// REAL
+/// DOUBLE
+/// BOOLEAN
+/// VARCHAR
+/// TIMESTAMP
+
+/// Special cases:
+/// NaN in Spark is handled differently from standard floating point semantics.
+/// It is considered larger than any other numeric values.
+
+std::shared_ptr<exec::VectorFunction> makeEqualTo(
+    const std::string& name,
+    const std::vector<exec::VectorFunctionArg>& inputArgs,
+    const core::QueryConfig& /*config*/);
+
+std::shared_ptr<exec::VectorFunction> makeLessThan(
+    const std::string& name,
+    const std::vector<exec::VectorFunctionArg>& inputArgs,
+    const core::QueryConfig& /*config*/);
+
+std::shared_ptr<exec::VectorFunction> makeGreaterThan(
+    const std::string& name,
+    const std::vector<exec::VectorFunctionArg>& inputArgs,
+    const core::QueryConfig& /*config*/);
+
+std::shared_ptr<exec::VectorFunction> makeLessThanOrEqual(
+    const std::string& name,
+    const std::vector<exec::VectorFunctionArg>& inputArgs,
+    const core::QueryConfig& /*config*/);
+
+std::shared_ptr<exec::VectorFunction> makeGreaterThanOrEqual(
+    const std::string& name,
+    const std::vector<exec::VectorFunctionArg>& inputArgs,
+    const core::QueryConfig& /*config*/);
+
+inline std::vector<std::shared_ptr<exec::FunctionSignature>>
+comparisonSignatures() {
+  return {exec::FunctionSignatureBuilder()
+              .typeVariable("T")
+              .returnType("boolean")
+              .argumentType("T")
+              .argumentType("T")
+              .build()};
+}
+
+template <typename T>
+struct BetweenFunction {
+  template <typename TInput>
+  FOLLY_ALWAYS_INLINE void call(
+      bool& result,
+      const TInput& value,
+      const TInput& low,
+      const TInput& high) {
+    result = value >= low && value <= high;
+  }
+};
+
 } // namespace facebook::velox::functions::sparksql
diff --git a/velox/functions/sparksql/RegisterCompare.cpp b/velox/functions/sparksql/RegisterCompare.cpp
index ef0f7a7f18de..f13655f1041e 100644
--- a/velox/functions/sparksql/RegisterCompare.cpp
+++ b/velox/functions/sparksql/RegisterCompare.cpp
@@ -16,18 +16,28 @@
 #include "velox/functions/sparksql/RegisterCompare.h"
 
 #include "velox/functions/lib/RegistrationHelpers.h"
-#include "velox/functions/prestosql/Comparisons.h"
+#include "velox/functions/sparksql/CompareFunctionsNullSafe.h"
+#include "velox/functions/sparksql/Comparisons.h"
 
 namespace facebook::velox::functions::sparksql {
 
 void registerCompareFunctions(const std::string& prefix) {
-  registerBinaryScalar<EqFunction, bool>({prefix + "equalto"});
-  registerBinaryScalar<NeqFunction, bool>({prefix + "notequalto"});
-  registerBinaryScalar<LtFunction, bool>({prefix + "lessthan"});
-  registerBinaryScalar<GtFunction, bool>({prefix + "greaterthan"});
-  registerBinaryScalar<LteFunction, bool>({prefix + "lessthanorequal"});
-  registerBinaryScalar<GteFunction, bool>({prefix + "greaterthanorequal"});
-
+  // Register compare functions
+  exec::registerStatefulVectorFunction(
+      prefix + "equalto", comparisonSignatures(), makeEqualTo);
+  exec::registerStatefulVectorFunction(
+      prefix + "lessthan", comparisonSignatures(), makeLessThan);
+  exec::registerStatefulVectorFunction(
+      prefix + "greaterthan", comparisonSignatures(), makeGreaterThan);
+  exec::registerStatefulVectorFunction(
+      prefix + "lessthanorequal", comparisonSignatures(), makeLessThanOrEqual);
+  exec::registerStatefulVectorFunction(
+      prefix + "greaterthanorequal",
+      comparisonSignatures(),
+      makeGreaterThanOrEqual);
+  // Compare nullsafe functions
+  exec::registerStatefulVectorFunction(
+      prefix + "equalnullsafe", equalNullSafeSignatures(), makeEqualNullSafe);
   registerFunction<BetweenFunction, bool, int8_t, int8_t, int8_t>(
       {prefix + "between"});
   registerFunction<BetweenFunction, bool, int16_t, int16_t, int16_t>(
@@ -40,6 +50,10 @@ void registerCompareFunctions(const std::string& prefix) {
       {prefix + "between"});
   registerFunction<BetweenFunction, bool, float, float, float>(
       {prefix + "between"});
+  registerFunction<BetweenFunction, bool, int64_t, int64_t, int64_t>(
+      {prefix + "between"});
+  registerFunction<BetweenFunction, bool, int128_t, int128_t, int128_t>(
+      {prefix + "between"});
 }
 
 } // namespace facebook::velox::functions::sparksql
diff --git a/velox/functions/sparksql/tests/CMakeLists.txt b/velox/functions/sparksql/tests/CMakeLists.txt
index 97291952ccfc..038ad0d83fc8 100644
--- a/velox/functions/sparksql/tests/CMakeLists.txt
+++ b/velox/functions/sparksql/tests/CMakeLists.txt
@@ -18,6 +18,7 @@ add_executable(
   ArraySortTest.cpp
   BitwiseTest.cpp
   CompareNullSafeTests.cpp
+  CompareTests.cpp
   DateTimeFunctionsTest.cpp
   ElementAtTest.cpp
   HashTest.cpp
diff --git a/velox/functions/sparksql/tests/CompareTests.cpp b/velox/functions/sparksql/tests/CompareTests.cpp
new file mode 100644
index 000000000000..f6f5e25d7292
--- /dev/null
+++ b/velox/functions/sparksql/tests/CompareTests.cpp
@@ -0,0 +1,298 @@
+/*
+ * 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 "velox/common/base/tests/GTestUtils.h"
+#include "velox/functions/sparksql/tests/SparkFunctionBaseTest.h"
+
+#include <velox/vector/SimpleVector.h>
+
+namespace facebook::velox::functions::sparksql::test {
+namespace {
+static constexpr double kInf = std::numeric_limits<double>::infinity();
+static constexpr float kInfF = std::numeric_limits<float>::infinity();
+static constexpr auto kNaN = std::numeric_limits<double>::quiet_NaN();
+class CompareTest : public SparkFunctionBaseTest {
+ protected:
+  template <typename T>
+  std::optional<bool> equaltonullsafe(std::optional<T> a, std::optional<T> b) {
+    return evaluateOnce<bool>("equalnullsafe(c0, c1)", a, b);
+  }
+
+  template <typename T>
+  std::optional<bool> equalto(std::optional<T> a, std::optional<T> b) {
+    return evaluateOnce<bool>("equalto(c0, c1)", a, b);
+  }
+
+  template <typename T>
+  std::optional<bool> lessthan(std::optional<T> a, std::optional<T> b) {
+    return evaluateOnce<bool>("lessthan(c0, c1)", a, b);
+  }
+
+  template <typename T>
+  std::optional<bool> lessthanorequal(std::optional<T> a, std::optional<T> b) {
+    return evaluateOnce<bool>("lessthanorequal(c0, c1)", a, b);
+  }
+
+  template <typename T>
+  std::optional<bool> greaterthan(std::optional<T> a, std::optional<T> b) {
+    return evaluateOnce<bool>("greaterthan(c0, c1)", a, b);
+  }
+
+  template <typename T>
+  std::optional<bool> greaterthanorequal(
+      std::optional<T> a,
+      std::optional<T> b) {
+    return evaluateOnce<bool>("greaterthanorequal(c0, c1)", a, b);
+  }
+};
+
+TEST_F(CompareTest, equaltonullsafe) {
+  EXPECT_EQ(equaltonullsafe<int64_t>(1, 1), true);
+  EXPECT_EQ(equaltonullsafe<int32_t>(1, 2), false);
+  EXPECT_EQ(equaltonullsafe<float>(std::nullopt, std::nullopt), true);
+  EXPECT_EQ(equaltonullsafe<std::string>(std::nullopt, "abcs"), false);
+  EXPECT_EQ(equaltonullsafe<std::string>(std::nullopt, std::nullopt), true);
+  EXPECT_EQ(equaltonullsafe<double>(1, std::nullopt), false);
+  EXPECT_EQ(equaltonullsafe<double>(kNaN, std::nullopt), false);
+  EXPECT_EQ(equaltonullsafe<double>(kNaN, 1), false);
+  EXPECT_EQ(equaltonullsafe<double>(kNaN, kNaN), true);
+}
+
+TEST_F(CompareTest, equalto) {
+  EXPECT_EQ(equalto<int64_t>(1, 1), true);
+  EXPECT_EQ(equalto<int32_t>(1, 2), false);
+  EXPECT_EQ(equalto<float>(std::nullopt, std::nullopt), std::nullopt);
+  EXPECT_EQ(equalto<std::string>(std::nullopt, "abcs"), std::nullopt);
+  EXPECT_EQ(equalto<std::string>(std::nullopt, std::nullopt), std::nullopt);
+  EXPECT_EQ(equalto<double>(1, std::nullopt), std::nullopt);
+  EXPECT_EQ(equalto<double>(kNaN, std::nullopt), std::nullopt);
+  EXPECT_EQ(equalto<double>(kNaN, 1), false);
+  EXPECT_EQ(equalto<double>(0, kNaN), false);
+  EXPECT_EQ(equalto<double>(kNaN, kNaN), true);
+  EXPECT_EQ(equalto<double>(kInf, kInf), true);
+  EXPECT_EQ(equalto<float>(kInfF, kInfF), true);
+  EXPECT_EQ(equalto<double>(kInf, 2.0), false);
+  EXPECT_EQ(equalto<double>(-kInf, 2.0), false);
+  EXPECT_EQ(equalto<float>(kInfF, 1.0), false);
+  EXPECT_EQ(equalto<float>(-kInfF, 1.0), false);
+  EXPECT_EQ(equalto<float>(kInfF, -kInfF), false);
+  EXPECT_EQ(equalto<double>(kInf, kNaN), false);
+}
+
+TEST_F(CompareTest, testdecimal) {
+  auto runAndCompare = [&](const std::string& exprStr,
+                           std::vector<VectorPtr>& input,
+                           VectorPtr expectedResult) {
+    auto actual = evaluate<SimpleVector<bool>>(exprStr, makeRowVector(input));
+    facebook::velox::test::assertEqualVectors(actual, expectedResult);
+  };
+  std::vector<VectorPtr> inputs = {
+      makeNullableShortDecimalFlatVector(
+          {1, std::nullopt, 3, -2, std::nullopt, 4}, DECIMAL(10, 5)),
+      makeNullableShortDecimalFlatVector(
+          {0, 2, 3, -3, std::nullopt, 5}, DECIMAL(10, 5))};
+  auto expected = makeNullableFlatVector<bool>(
+      {true, std::nullopt, false, true, std::nullopt, false});
+  runAndCompare(fmt::format("{}(c0, c1)", "greaterthan"), inputs, expected);
+  std::vector<VectorPtr> longDecimalsInputs = {
+      makeNullableLongDecimalFlatVector(
+          {DecimalUtil::kLongDecimalMax,
+           std::nullopt,
+           3,
+           DecimalUtil::kLongDecimalMin + 1,
+           std::nullopt,
+           4},
+          DECIMAL(38, 5)),
+      makeNullableLongDecimalFlatVector(
+          {DecimalUtil::kLongDecimalMax - 1,
+           2,
+           3,
+           DecimalUtil::kLongDecimalMin,
+           std::nullopt,
+           5},
+          DECIMAL(38, 5))};
+  auto expectedGteLte = makeNullableFlatVector<bool>(
+      {true, std::nullopt, true, true, std::nullopt, false});
+  runAndCompare(
+      fmt::format("{}(c1, c0)", "lessthanorequal"),
+      longDecimalsInputs,
+      expectedGteLte);
+
+  // Test with different data types.
+  std::vector<VectorPtr> invalidInputs = {
+      makeNullableShortDecimalFlatVector({1}, DECIMAL(10, 5)),
+      makeNullableShortDecimalFlatVector({1}, DECIMAL(10, 4))};
+  auto invalidResult = makeNullableFlatVector<bool>({true});
+  VELOX_ASSERT_THROW(
+      runAndCompare(
+          fmt::format("{}(c1, c0)", "equalto"), invalidInputs, invalidResult),
+      "Scalar function signature is not supported: "
+      "equalto(DECIMAL(10,4), DECIMAL(10,5))");
+}
+
+TEST_F(CompareTest, testdictionary) {
+  // Identity mapping, however this will result in non-simd path.
+  auto makeDictionary = [&](const VectorPtr& base) {
+    auto indices = makeIndices(base->size(), [](auto row) { return row; });
+    return wrapInDictionary(indices, base->size(), base);
+  };
+
+  auto makeConstantDic = [&](const VectorPtr& base) {
+    return BaseVector::wrapInConstant(base->size(), 0, base);
+  };
+  // lhs: 0, null, 2, null, 4
+
+  auto lhs = makeFlatVector<int16_t>(
+      5, [](auto row) { return row; }, nullEvery(2, 1));
+  auto rhs = makeFlatVector<int16_t>({1, 0, 3, 0, 5});
+  auto lhsVector = makeDictionary(lhs);
+  auto rhsVector = makeDictionary(rhs);
+
+  auto rowVector = makeRowVector({lhsVector, rhsVector});
+  auto result = evaluate<SimpleVector<bool>>(
+      fmt::format("{}(c0, c1)", "greaterthan"), rowVector);
+  // result : false, null, false, null, false
+  facebook::velox::test::assertEqualVectors(
+      result,
+      makeFlatVector<bool>(
+          5, [](auto row) { return false; }, nullEvery(2, 1)));
+  auto constVector = makeConstant(100, 5);
+  auto testConstVector =
+      makeRowVector({makeDictionary(lhs), makeConstantDic(constVector)});
+  // lhs: 0, null, 2, null, 4
+  // rhs: const 100
+  // lessthanorequal result : true, null, true, null, true
+  auto constResult = evaluate<SimpleVector<bool>>(
+      fmt::format("{}(c0, c1)", "lessthanorequal"), testConstVector);
+  facebook::velox::test::assertEqualVectors(
+      constResult,
+      makeFlatVector<bool>(
+          5, [](auto row) { return true; }, nullEvery(2, 1)));
+  // lhs: const 100
+  // rhs: 0, null, 2, null, 4
+  // greaterthanorequal result : true, null, true, null, true
+  auto testConstVector1 =
+      makeRowVector({makeConstantDic(constVector), makeDictionary(lhs)});
+  auto constResult1 = evaluate<SimpleVector<bool>>(
+      fmt::format("{}(c0, c1)", "greaterthanorequal"), testConstVector1);
+  facebook::velox::test::assertEqualVectors(
+      constResult1,
+      makeFlatVector<bool>(
+          5, [](auto row) { return true; }, nullEvery(2, 1)));
+}
+
+TEST_F(CompareTest, testflat) {
+  auto vector0 = makeFlatVector<int32_t>({0, 1, 2, 3});
+  auto vector1 = makeFlatVector<int32_t>(
+      4, [](auto row) { return row + 1; }, nullEvery(2));
+
+  auto expectedResult = makeFlatVector<bool>(
+      4, [](auto row) { return true; }, nullEvery(2));
+  auto actualResult = evaluate<SimpleVector<bool>>(
+      "lessthan(c0, c1)", makeRowVector({vector0, vector1}));
+  facebook::velox::test::assertEqualVectors(expectedResult, actualResult);
+}
+
+TEST_F(CompareTest, lessthan) {
+  EXPECT_EQ(lessthan<int64_t>(1, 1), false);
+  EXPECT_EQ(lessthan<int32_t>(1, 2), true);
+  EXPECT_EQ(lessthan<float>(std::nullopt, std::nullopt), std::nullopt);
+  EXPECT_EQ(lessthan<std::string>(std::nullopt, "abcs"), std::nullopt);
+  EXPECT_EQ(lessthan<std::string>(std::nullopt, std::nullopt), std::nullopt);
+  EXPECT_EQ(lessthan<double>(1, std::nullopt), std::nullopt);
+  EXPECT_EQ(lessthan<double>(kNaN, std::nullopt), std::nullopt);
+  EXPECT_EQ(lessthan<double>(kNaN, 1), false);
+  EXPECT_EQ(lessthan<double>(0, kNaN), true);
+  EXPECT_EQ(lessthan<double>(kNaN, kNaN), false);
+  EXPECT_EQ(lessthan<double>(kInf, kInf), false);
+  EXPECT_EQ(lessthan<float>(kInfF, kInfF), false);
+  EXPECT_EQ(lessthan<double>(kInf, 2.0), false);
+  EXPECT_EQ(lessthan<double>(-kInf, 2.0), true);
+  EXPECT_EQ(lessthan<float>(kInfF, 1.0), false);
+  EXPECT_EQ(lessthan<float>(-kInfF, 1.0), true);
+  EXPECT_EQ(lessthan<float>(kInfF, -kInfF), false);
+  EXPECT_EQ(lessthan<double>(kInf, kNaN), true);
+}
+
+TEST_F(CompareTest, lessthanorequal) {
+  EXPECT_EQ(lessthanorequal<int64_t>(1, 1), true);
+  EXPECT_EQ(lessthanorequal<int32_t>(1, 2), true);
+  EXPECT_EQ(lessthanorequal<float>(std::nullopt, std::nullopt), std::nullopt);
+  EXPECT_EQ(lessthanorequal<std::string>(std::nullopt, "abcs"), std::nullopt);
+  EXPECT_EQ(
+      lessthanorequal<std::string>(std::nullopt, std::nullopt), std::nullopt);
+  EXPECT_EQ(lessthanorequal<double>(1, std::nullopt), std::nullopt);
+  EXPECT_EQ(lessthanorequal<double>(kNaN, std::nullopt), std::nullopt);
+  EXPECT_EQ(lessthanorequal<double>(kNaN, 1), false);
+  EXPECT_EQ(lessthanorequal<double>(0, kNaN), true);
+  EXPECT_EQ(lessthanorequal<double>(kNaN, kNaN), true);
+  EXPECT_EQ(lessthanorequal<double>(kInf, kInf), true);
+  EXPECT_EQ(lessthanorequal<float>(kInfF, kInfF), true);
+  EXPECT_EQ(lessthanorequal<double>(kInf, 2.0), false);
+  EXPECT_EQ(lessthanorequal<double>(-kInf, 2.0), true);
+  EXPECT_EQ(lessthanorequal<float>(kInfF, 1.0), false);
+  EXPECT_EQ(lessthanorequal<float>(-kInfF, 1.0), true);
+  EXPECT_EQ(lessthanorequal<float>(kInfF, -kInfF), false);
+  EXPECT_EQ(lessthanorequal<double>(kInf, kNaN), true);
+}
+
+TEST_F(CompareTest, greaterthan) {
+  EXPECT_EQ(greaterthan<int64_t>(1, 1), false);
+  EXPECT_EQ(greaterthan<int32_t>(1, 2), false);
+  EXPECT_EQ(greaterthan<float>(std::nullopt, std::nullopt), std::nullopt);
+  EXPECT_EQ(greaterthan<std::string>(std::nullopt, "abcs"), std::nullopt);
+  EXPECT_EQ(greaterthan<std::string>(std::nullopt, std::nullopt), std::nullopt);
+  EXPECT_EQ(greaterthan<double>(1, std::nullopt), std::nullopt);
+  EXPECT_EQ(greaterthan<double>(kNaN, std::nullopt), std::nullopt);
+  EXPECT_EQ(greaterthan<double>(kNaN, 1), true);
+  EXPECT_EQ(greaterthan<double>(0, kNaN), false);
+  EXPECT_EQ(greaterthan<double>(kNaN, kNaN), false);
+  EXPECT_EQ(greaterthan<double>(kInf, kInf), false);
+  EXPECT_EQ(greaterthan<float>(kInfF, kInfF), false);
+  EXPECT_EQ(greaterthan<double>(kInf, 2.0), true);
+  EXPECT_EQ(greaterthan<double>(-kInf, 2.0), false);
+  EXPECT_EQ(greaterthan<float>(kInfF, 1.0), true);
+  EXPECT_EQ(greaterthan<float>(-kInfF, 1.0), false);
+  EXPECT_EQ(greaterthan<float>(kInfF, -kInfF), true);
+  EXPECT_EQ(greaterthan<float>(kInf, kNaN), false);
+}
+
+TEST_F(CompareTest, greaterthanorequal) {
+  EXPECT_EQ(greaterthanorequal<int64_t>(1, 1), true);
+  EXPECT_EQ(greaterthanorequal<int32_t>(1, 2), false);
+  EXPECT_EQ(
+      greaterthanorequal<float>(std::nullopt, std::nullopt), std::nullopt);
+  EXPECT_EQ(
+      greaterthanorequal<std::string>(std::nullopt, "abcs"), std::nullopt);
+  EXPECT_EQ(
+      greaterthanorequal<std::string>(std::nullopt, std::nullopt),
+      std::nullopt);
+  EXPECT_EQ(greaterthanorequal<double>(1, std::nullopt), std::nullopt);
+  EXPECT_EQ(greaterthanorequal<double>(kNaN, std::nullopt), std::nullopt);
+  EXPECT_EQ(greaterthanorequal<double>(kNaN, 1), true);
+  EXPECT_EQ(greaterthanorequal<double>(0, kNaN), false);
+  EXPECT_EQ(greaterthanorequal<double>(kNaN, kNaN), true);
+  EXPECT_EQ(greaterthanorequal<double>(kInf, kInf), true);
+  EXPECT_EQ(greaterthanorequal<float>(kInfF, kInfF), true);
+  EXPECT_EQ(greaterthanorequal<double>(kInf, 2.0), true);
+  EXPECT_EQ(greaterthanorequal<double>(-kInf, 2.0), false);
+  EXPECT_EQ(greaterthanorequal<float>(kInfF, 1.0), true);
+  EXPECT_EQ(greaterthanorequal<float>(-kInfF, 1.0), false);
+  EXPECT_EQ(greaterthanorequal<float>(kInfF, -kInfF), true);
+  EXPECT_EQ(greaterthanorequal<float>(kInf, kNaN), false);
+}
+} // namespace
+}; // namespace facebook::velox::functions::sparksql::test