diff --git a/velox/functions/prestosql/aggregates/CMakeLists.txt b/velox/functions/prestosql/aggregates/CMakeLists.txt
index 30872f83a442..30f2c60b65d5 100644
--- a/velox/functions/prestosql/aggregates/CMakeLists.txt
+++ b/velox/functions/prestosql/aggregates/CMakeLists.txt
@@ -33,6 +33,7 @@ add_library(
   MapAggregateBase.cpp
   MapUnionAggregate.cpp
   MinMaxAggregates.cpp
+  MinMaxAggregates.h
   MinMaxByAggregates.cpp
   CountAggregate.cpp
   CountAggregate.h
diff --git a/velox/functions/prestosql/aggregates/MinMaxAggregates.cpp b/velox/functions/prestosql/aggregates/MinMaxAggregates.cpp
index 19d76d164fb4..dfa891ea3857 100644
--- a/velox/functions/prestosql/aggregates/MinMaxAggregates.cpp
+++ b/velox/functions/prestosql/aggregates/MinMaxAggregates.cpp
@@ -14,13 +14,8 @@
  * limitations under the License.
  */
 
-#include <limits>
-#include "velox/exec/Aggregate.h"
-#include "velox/exec/AggregationHook.h"
-#include "velox/expression/FunctionSignature.h"
+#include "velox/functions/prestosql/aggregates/MinMaxAggregates.h"
 #include "velox/functions/prestosql/aggregates/AggregateNames.h"
-#include "velox/functions/prestosql/aggregates/SimpleNumericAggregate.h"
-#include "velox/functions/prestosql/aggregates/SingleValueAccumulator.h"
 
 namespace facebook::velox::aggregate::prestosql {
 
diff --git a/velox/functions/prestosql/aggregates/MinMaxAggregates.h b/velox/functions/prestosql/aggregates/MinMaxAggregates.h
new file mode 100644
index 000000000000..56b500bd61a6
--- /dev/null
+++ b/velox/functions/prestosql/aggregates/MinMaxAggregates.h
@@ -0,0 +1,498 @@
+/*
+ * 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 <limits>
+#include "velox/exec/Aggregate.h"
+#include "velox/exec/AggregationHook.h"
+#include "velox/expression/FunctionSignature.h"
+#include "velox/functions/prestosql/aggregates/AggregateNames.h"
+#include "velox/functions/prestosql/aggregates/SimpleNumericAggregate.h"
+#include "velox/functions/prestosql/aggregates/SingleValueAccumulator.h"
+
+namespace facebook::velox::aggregate {
+
+template <typename T>
+struct MinMaxTrait : public std::numeric_limits<T> {};
+
+template <>
+struct MinMaxTrait<Timestamp> {
+  static constexpr Timestamp min() {
+    return Timestamp(MinMaxTrait<int64_t>::min(), MinMaxTrait<uint64_t>::min());
+  }
+
+  static constexpr Timestamp max() {
+    return Timestamp(MinMaxTrait<int64_t>::max(), MinMaxTrait<uint64_t>::max());
+  }
+};
+
+template <>
+struct MinMaxTrait<Date> {
+  static constexpr Date min() {
+    return Date(std::numeric_limits<int32_t>::min());
+  }
+
+  static constexpr Date max() {
+    return Date(std::numeric_limits<int32_t>::max());
+  }
+};
+
+template <typename T>
+class MinMaxAggregate : public SimpleNumericAggregate<T, T, T> {
+  using BaseAggregate = SimpleNumericAggregate<T, T, T>;
+
+ public:
+  explicit MinMaxAggregate(TypePtr resultType) : BaseAggregate(resultType) {}
+
+  int32_t accumulatorFixedWidthSize() const override {
+    return sizeof(T);
+  }
+
+  void extractValues(char** groups, int32_t numGroups, VectorPtr* result)
+      override {
+    BaseAggregate::template doExtractValues<T>(
+        groups, numGroups, result, [&](char* group) {
+          return *BaseAggregate::Aggregate::template value<T>(group);
+        });
+  }
+
+  void extractAccumulators(char** groups, int32_t numGroups, VectorPtr* result)
+      override {
+    BaseAggregate::template doExtractValues<T>(
+        groups, numGroups, result, [&](char* group) {
+          return *BaseAggregate::Aggregate::template value<T>(group);
+        });
+  }
+};
+
+// Truncate timestamps to milliseconds precision.
+template <>
+void MinMaxAggregate<Timestamp>::extractValues(
+    char** groups,
+    int32_t numGroups,
+    VectorPtr* result) {
+  BaseAggregate::template doExtractValues<Timestamp>(
+      groups, numGroups, result, [&](char* group) {
+        auto ts = *BaseAggregate::Aggregate::template value<Timestamp>(group);
+        return Timestamp::fromMillis(ts.toMillis());
+      });
+}
+
+template <typename T>
+class MaxAggregate : public MinMaxAggregate<T> {
+  using BaseAggregate = SimpleNumericAggregate<T, T, T>;
+
+ public:
+  explicit MaxAggregate(TypePtr resultType) : MinMaxAggregate<T>(resultType) {}
+
+  void initializeNewGroups(
+      char** groups,
+      folly::Range<const vector_size_t*> indices) override {
+    exec::Aggregate::setAllNulls(groups, indices);
+    for (auto i : indices) {
+      *exec::Aggregate::value<T>(groups[i]) = kInitialValue_;
+    }
+  }
+
+  void addRawInput(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    if (mayPushdown && args[0]->isLazy()) {
+      BaseAggregate::template pushdown<MinMaxHook<T, false>>(
+          groups, rows, args[0]);
+      return;
+    }
+    BaseAggregate::template updateGroups<true, T>(
+        groups,
+        rows,
+        args[0],
+        [](T& result, T value) {
+          if (result < value) {
+            result = value;
+          }
+        },
+        mayPushdown);
+  }
+
+  void addIntermediateResults(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    addRawInput(groups, rows, args, mayPushdown);
+  }
+
+  void addSingleGroupRawInput(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    BaseAggregate::updateOneGroup(
+        group,
+        rows,
+        args[0],
+        [](T& result, T value) { result = result > value ? result : value; },
+        [](T& result, T value, int /* unused */) { result = value; },
+        mayPushdown,
+        kInitialValue_);
+  }
+
+  void addSingleGroupIntermediateResults(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    addSingleGroupRawInput(group, rows, args, mayPushdown);
+  }
+
+ private:
+  static constexpr T kInitialValue_{MinMaxTrait<T>::min()};
+};
+
+template <typename T>
+class MinAggregate : public MinMaxAggregate<T> {
+  using BaseAggregate = SimpleNumericAggregate<T, T, T>;
+
+ public:
+  explicit MinAggregate(TypePtr resultType) : MinMaxAggregate<T>(resultType) {}
+
+  void initializeNewGroups(
+      char** groups,
+      folly::Range<const vector_size_t*> indices) override {
+    exec::Aggregate::setAllNulls(groups, indices);
+    for (auto i : indices) {
+      *exec::Aggregate::value<T>(groups[i]) = kInitialValue_;
+    }
+  }
+
+  void addRawInput(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    if (mayPushdown && args[0]->isLazy()) {
+      BaseAggregate::template pushdown<MinMaxHook<T, true>>(
+          groups, rows, args[0]);
+      return;
+    }
+    BaseAggregate::template updateGroups<true, T>(
+        groups,
+        rows,
+        args[0],
+        [](T& result, T value) {
+          if (result > value) {
+            result = value;
+          }
+        },
+        mayPushdown);
+  }
+
+  void addIntermediateResults(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    addRawInput(groups, rows, args, mayPushdown);
+  }
+
+  void addSingleGroupRawInput(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    BaseAggregate::updateOneGroup(
+        group,
+        rows,
+        args[0],
+        [](T& result, T value) { result = result < value ? result : value; },
+        [](T& result, T value, int /* unused */) { result = value; },
+        mayPushdown,
+        kInitialValue_);
+  }
+
+  void addSingleGroupIntermediateResults(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    addSingleGroupRawInput(group, rows, args, mayPushdown);
+  }
+
+ private:
+  static constexpr T kInitialValue_{MinMaxTrait<T>::max()};
+};
+
+class NonNumericMinMaxAggregateBase : public exec::Aggregate {
+ public:
+  explicit NonNumericMinMaxAggregateBase(const TypePtr& resultType)
+      : exec::Aggregate(resultType) {}
+
+  int32_t accumulatorFixedWidthSize() const override {
+    return sizeof(SingleValueAccumulator);
+  }
+
+  void initializeNewGroups(
+      char** groups,
+      folly::Range<const vector_size_t*> indices) override {
+    exec::Aggregate::setAllNulls(groups, indices);
+    for (auto i : indices) {
+      new (groups[i] + offset_) SingleValueAccumulator();
+    }
+  }
+
+  void finalize(char** /* groups */, int32_t /* numGroups */) override {
+    // Nothing to do
+  }
+
+  void extractValues(char** groups, int32_t numGroups, VectorPtr* result)
+      override {
+    VELOX_CHECK(result);
+    (*result)->resize(numGroups);
+
+    uint64_t* rawNulls = nullptr;
+    if ((*result)->mayHaveNulls()) {
+      BufferPtr nulls = (*result)->mutableNulls((*result)->size());
+      rawNulls = nulls->asMutable<uint64_t>();
+    }
+
+    for (auto i = 0; i < numGroups; ++i) {
+      char* group = groups[i];
+      auto accumulator = value<SingleValueAccumulator>(group);
+      if (!accumulator->hasValue()) {
+        (*result)->setNull(i, true);
+      } else {
+        if (rawNulls) {
+          bits::clearBit(rawNulls, i);
+        }
+        accumulator->read(*result, i);
+      }
+    }
+  }
+
+  void extractAccumulators(char** groups, int32_t numGroups, VectorPtr* result)
+      override {
+    // partial and final aggregations are the same
+    extractValues(groups, numGroups, result);
+  }
+
+  void destroy(folly::Range<char**> groups) override {
+    for (auto group : groups) {
+      value<SingleValueAccumulator>(group)->destroy(allocator_);
+    }
+  }
+
+ protected:
+  template <typename TCompareTest>
+  void doUpdate(
+      char** groups,
+      const SelectivityVector& rows,
+      const VectorPtr& arg,
+      TCompareTest compareTest) {
+    DecodedVector decoded(*arg, rows, true);
+    auto indices = decoded.indices();
+    auto baseVector = decoded.base();
+
+    if (decoded.isConstantMapping() && decoded.isNullAt(0)) {
+      // nothing to do; all values are nulls
+      return;
+    }
+
+    rows.applyToSelected([&](vector_size_t i) {
+      if (decoded.isNullAt(i)) {
+        return;
+      }
+      auto accumulator = value<SingleValueAccumulator>(groups[i]);
+      if (!accumulator->hasValue() ||
+          compareTest(accumulator->compare(decoded, i))) {
+        accumulator->write(baseVector, indices[i], allocator_);
+      }
+    });
+  }
+
+  template <typename TCompareTest>
+  void doUpdateSingleGroup(
+      char* group,
+      const SelectivityVector& rows,
+      const VectorPtr& arg,
+      TCompareTest compareTest) {
+    DecodedVector decoded(*arg, rows, true);
+    auto indices = decoded.indices();
+    auto baseVector = decoded.base();
+
+    if (decoded.isConstantMapping()) {
+      if (decoded.isNullAt(0)) {
+        // nothing to do; all values are nulls
+        return;
+      }
+
+      auto accumulator = value<SingleValueAccumulator>(group);
+      if (!accumulator->hasValue() ||
+          compareTest(accumulator->compare(decoded, 0))) {
+        accumulator->write(baseVector, indices[0], allocator_);
+      }
+      return;
+    }
+
+    auto accumulator = value<SingleValueAccumulator>(group);
+    rows.applyToSelected([&](vector_size_t i) {
+      if (decoded.isNullAt(i)) {
+        return;
+      }
+      if (!accumulator->hasValue() ||
+          compareTest(accumulator->compare(decoded, i))) {
+        accumulator->write(baseVector, indices[i], allocator_);
+      }
+    });
+  }
+};
+
+class NonNumericMaxAggregate : public NonNumericMinMaxAggregateBase {
+ public:
+  explicit NonNumericMaxAggregate(const TypePtr& resultType)
+      : NonNumericMinMaxAggregateBase(resultType) {}
+
+  void addRawInput(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool /*mayPushdown*/) override {
+    doUpdate(groups, rows, args[0], [](int32_t compareResult) {
+      return compareResult < 0;
+    });
+  }
+
+  void addIntermediateResults(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    addRawInput(groups, rows, args, mayPushdown);
+  }
+
+  void addSingleGroupRawInput(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool /*mayPushdown*/) override {
+    doUpdateSingleGroup(group, rows, args[0], [](int32_t compareResult) {
+      return compareResult < 0;
+    });
+  }
+
+  void addSingleGroupIntermediateResults(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    addSingleGroupRawInput(group, rows, args, mayPushdown);
+  }
+};
+
+class NonNumericMinAggregate : public NonNumericMinMaxAggregateBase {
+ public:
+  explicit NonNumericMinAggregate(const TypePtr& resultType)
+      : NonNumericMinMaxAggregateBase(resultType) {}
+
+  void addRawInput(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool /*mayPushdown*/) override {
+    doUpdate(groups, rows, args[0], [](int32_t compareResult) {
+      return compareResult > 0;
+    });
+  }
+
+  void addIntermediateResults(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    addRawInput(groups, rows, args, mayPushdown);
+  }
+
+  void addSingleGroupRawInput(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool /*mayPushdown*/) override {
+    doUpdateSingleGroup(group, rows, args[0], [](int32_t compareResult) {
+      return compareResult > 0;
+    });
+  }
+
+  void addSingleGroupIntermediateResults(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    addSingleGroupRawInput(group, rows, args, mayPushdown);
+  }
+};
+
+template <template <typename T> class TNumeric, typename TNonNumeric>
+bool registerMinMaxAggregate(const std::string& name) {
+  std::vector<std::shared_ptr<exec::AggregateFunctionSignature>> signatures;
+  signatures.push_back(exec::AggregateFunctionSignatureBuilder()
+                           .typeVariable("T")
+                           .returnType("T")
+                           .intermediateType("T")
+                           .argumentType("T")
+                           .build());
+
+  return exec::registerAggregateFunction(
+      name,
+      std::move(signatures),
+      [name](
+          core::AggregationNode::Step step,
+          std::vector<TypePtr> argTypes,
+          const TypePtr& resultType) -> std::unique_ptr<exec::Aggregate> {
+        VELOX_CHECK_EQ(argTypes.size(), 1, "{} takes only one argument", name);
+        auto inputType = argTypes[0];
+        switch (inputType->kind()) {
+          case TypeKind::TINYINT:
+            return std::make_unique<TNumeric<int8_t>>(resultType);
+          case TypeKind::SMALLINT:
+            return std::make_unique<TNumeric<int16_t>>(resultType);
+          case TypeKind::INTEGER:
+            return std::make_unique<TNumeric<int32_t>>(resultType);
+          case TypeKind::BIGINT:
+            return std::make_unique<TNumeric<int64_t>>(resultType);
+          case TypeKind::REAL:
+            return std::make_unique<TNumeric<float>>(resultType);
+          case TypeKind::DOUBLE:
+            return std::make_unique<TNumeric<double>>(resultType);
+          case TypeKind::TIMESTAMP:
+            return std::make_unique<TNumeric<Timestamp>>(resultType);
+          case TypeKind::DATE:
+            return std::make_unique<TNumeric<Date>>(resultType);
+          case TypeKind::VARCHAR:
+          case TypeKind::ARRAY:
+          case TypeKind::MAP:
+          case TypeKind::ROW:
+            return std::make_unique<TNonNumeric>(inputType);
+          default:
+            VELOX_CHECK(
+                false,
+                "Unknown input type for {} aggregation {}",
+                name,
+                inputType->kindName());
+        }
+      });
+}
+
+} // namespace facebook::velox::aggregate
diff --git a/velox/substrait/CMakeLists.txt b/velox/substrait/CMakeLists.txt
index 612ae8f0f7a6..f6acbdcfc8d1 100644
--- a/velox/substrait/CMakeLists.txt
+++ b/velox/substrait/CMakeLists.txt
@@ -43,6 +43,7 @@ add_custom_command(
 add_custom_target(substrait_proto ALL DEPENDS ${PROTO_OUTPUT_FILES})
 add_dependencies(substrait_proto protobuf::libprotobuf)
 
+
 set(SRCS
     ${PROTO_SRCS}
     SubstraitParser.cpp
@@ -61,7 +62,7 @@ add_library(velox_substrait_plan_converter ${SRCS})
 target_include_directories(velox_substrait_plan_converter
                            PUBLIC ${PROTO_OUTPUT_DIR})
 target_link_libraries(velox_substrait_plan_converter velox_connector
-                      velox_dwio_dwrf_common)
+                      velox_dwio_dwrf_common velox_functions_spark)
 
 if(${VELOX_BUILD_TESTING})
   add_subdirectory(tests)
diff --git a/velox/substrait/SubstraitParser.cpp b/velox/substrait/SubstraitParser.cpp
index 51aa7e62e0ce..b255635b0f6a 100644
--- a/velox/substrait/SubstraitParser.cpp
+++ b/velox/substrait/SubstraitParser.cpp
@@ -142,14 +142,12 @@ std::shared_ptr<SubstraitParser::SubstraitType> SubstraitParser::parseType(
 std::vector<std::shared_ptr<SubstraitParser::SubstraitType>>
 SubstraitParser::parseNamedStruct(const ::substrait::NamedStruct& namedStruct) {
   // Nte that "names" are not used.
-
   // Parse Struct.
-  const auto& substraitStruct = namedStruct.struct_();
-  const auto& substraitTypes = substraitStruct.types();
-  std::vector<std::shared_ptr<SubstraitParser::SubstraitType>>
-      substraitTypeList;
-  substraitTypeList.reserve(substraitTypes.size());
-  for (const auto& type : substraitTypes) {
+  const auto& sStruct = namedStruct.struct_();
+  const auto& sTypes = sStruct.types();
+  std::vector<std::shared_ptr<SubstraitType>> substraitTypeList;
+  substraitTypeList.reserve(sTypes.size());
+  for (const auto& type : sTypes) {
     substraitTypeList.emplace_back(parseType(type));
   }
   return substraitTypeList;
diff --git a/velox/substrait/SubstraitParser.h b/velox/substrait/SubstraitParser.h
index c0ff587d1a9f..4dbf652f8a60 100644
--- a/velox/substrait/SubstraitParser.h
+++ b/velox/substrait/SubstraitParser.h
@@ -37,8 +37,8 @@ class SubstraitParser {
     bool nullable;
   };
 
-  /// Parse Substrait NamedStruct.
-  std::vector<std::shared_ptr<SubstraitParser::SubstraitType>> parseNamedStruct(
+  /// Used to parse Substrait NamedStruct.
+  std::vector<std::shared_ptr<SubstraitType>> parseNamedStruct(
       const ::substrait::NamedStruct& namedStruct);
 
   /// Parse Substrait Type.
@@ -102,6 +102,8 @@ class SubstraitParser {
       {"subtract", "minus"},
       {"modulus", "mod"},
       {"not_equal", "neq"},
+      {"ends_with", "endswith"},
+      {"starts_with", "startswith"},
       {"equal", "eq"}};
 };
 
diff --git a/velox/substrait/SubstraitToVeloxExpr.cpp b/velox/substrait/SubstraitToVeloxExpr.cpp
index d818806dac28..19e4f039de77 100644
--- a/velox/substrait/SubstraitToVeloxExpr.cpp
+++ b/velox/substrait/SubstraitToVeloxExpr.cpp
@@ -16,8 +16,9 @@
 
 #include "velox/substrait/SubstraitToVeloxExpr.h"
 #include "velox/substrait/TypeUtils.h"
-#include "velox/vector/FlatVector.h"
 
+#include "velox/substrait/VectorCreater.h"
+#include "velox/vector/FlatVector.h"
 using namespace facebook::velox;
 namespace {
 // Get values for the different supported types.
@@ -178,6 +179,28 @@ SubstraitVeloxExprConverter::toVeloxExpr(
   }
 }
 
+std::shared_ptr<const core::ITypedExpr>
+SubstraitVeloxExprConverter::toAliasExpr(
+    const std::vector<std::shared_ptr<const core::ITypedExpr>>& params) {
+  VELOX_CHECK(params.size() == 1, "Alias expects one parameter.");
+  // Alias is omitted due to name change is not needed.
+  return params[0];
+}
+
+std::shared_ptr<const core::ITypedExpr>
+SubstraitVeloxExprConverter::toIsNotNullExpr(
+    const std::vector<std::shared_ptr<const core::ITypedExpr>>& params,
+    const TypePtr& outputType) {
+  // Convert is_not_null to not(is_null).
+  auto isNullExpr = std::make_shared<const core::CallTypedExpr>(
+      outputType, std::move(params), "is_null");
+  std::vector<std::shared_ptr<const core::ITypedExpr>> notParams;
+  notParams.reserve(1);
+  notParams.emplace_back(isNullExpr);
+  return std::make_shared<const core::CallTypedExpr>(
+      outputType, std::move(notParams), "not");
+}
+
 std::shared_ptr<const core::ITypedExpr>
 SubstraitVeloxExprConverter::toVeloxExpr(
     const ::substrait::Expression::ScalarFunction& substraitFunc,
@@ -192,14 +215,12 @@ SubstraitVeloxExprConverter::toVeloxExpr(
   const auto& veloxType =
       toVeloxType(subParser_->parseType(sFunc.output_type())->type);
 
-  // Omit alias because because name change is not needed.
   if (veloxFunction == "alias") {
-    if (params.size() != 1) {
-      VELOX_FAIL("Alias expects one parameter.");
-    }
-    return params[0];
+    return toAliasExpr(params);
+  }
+  if (veloxFunction == "is_not_null") {
+    return toIsNotNullExpr(params, veloxType);
   }
-
   return std::make_shared<const core::CallTypedExpr>(
       toVeloxType(typeName), std::move(params), veloxFunction);
 }
@@ -240,6 +261,35 @@ SubstraitVeloxExprConverter::toVeloxExpr(
       return std::make_shared<core::ConstantTypedExpr>(
           veloxType, variant::null(veloxType->kind()));
     }
+    case ::substrait::Expression_Literal::LiteralTypeCase::kList: {
+      // List is used in 'in' expression. Will wrap a constant
+      // vector with an array vector inside to create the constant expression.
+      std::vector<variant> variants;
+      variants.reserve(sLit.list().values().size());
+      VELOX_CHECK(
+          sLit.list().values().size() > 0,
+          "List should have at least one item.");
+      std::optional<TypePtr> literalType = std::nullopt;
+      for (const auto& literal : sLit.list().values()) {
+        auto typedVariant = toTypedVariant(literal);
+        if (!literalType.has_value()) {
+          literalType = typedVariant->variantType;
+        }
+        variants.emplace_back(typedVariant->veloxVariant);
+      }
+      VELOX_CHECK(literalType.has_value(), "Type expected.");
+      // Create flat vector from the variants.
+      VectorPtr vector =
+          setVectorFromVariants(literalType.value(), variants, pool_);
+      // Create array vector from the flat vector.
+      ArrayVectorPtr arrayVector =
+          toArrayVector(literalType.value(), vector, pool_);
+      // Wrap the array vector into constant vector.
+      auto constantVector = BaseVector::wrapInConstant(1, 0, arrayVector);
+      auto constantExpr =
+          std::make_shared<core::ConstantTypedExpr>(constantVector);
+      return constantExpr;
+    }
     case ::substrait::Expression_Literal::LiteralTypeCase::kVarChar:
       return std::make_shared<core::ConstantTypedExpr>(
           variant(substraitLit.var_char().value()));
@@ -338,27 +388,66 @@ SubstraitVeloxExprConverter::toVeloxExpr(
 
 std::shared_ptr<const core::ITypedExpr>
 SubstraitVeloxExprConverter::toVeloxExpr(
-    const ::substrait::Expression& substraitExpr,
+    const ::substrait::Expression& sExpr,
     const RowTypePtr& inputType) {
   std::shared_ptr<const core::ITypedExpr> veloxExpr;
-  auto typeCase = substraitExpr.rex_type_case();
+  auto typeCase = sExpr.rex_type_case();
   switch (typeCase) {
     case ::substrait::Expression::RexTypeCase::kLiteral:
-      return toVeloxExpr(substraitExpr.literal());
+      return toVeloxExpr(sExpr.literal());
     case ::substrait::Expression::RexTypeCase::kScalarFunction:
-      return toVeloxExpr(substraitExpr.scalar_function(), inputType);
+      return toVeloxExpr(sExpr.scalar_function(), inputType);
     case ::substrait::Expression::RexTypeCase::kSelection:
-      return toVeloxExpr(substraitExpr.selection(), inputType);
+      return toVeloxExpr(sExpr.selection(), inputType);
     case ::substrait::Expression::RexTypeCase::kCast:
-      return toVeloxExpr(substraitExpr.cast(), inputType);
+      return toVeloxExpr(sExpr.cast(), inputType);
     case ::substrait::Expression::RexTypeCase::kIfThen:
-      return toVeloxExpr(substraitExpr.if_then(), inputType);
+      return toVeloxExpr(sExpr.if_then(), inputType);
+    case ::substrait::Expression::RexTypeCase::kSingularOrList:
+      return toVeloxExpr(sExpr.singular_or_list(), inputType);
     default:
       VELOX_NYI(
           "Substrait conversion not supported for Expression '{}'", typeCase);
   }
 }
 
+std::shared_ptr<const core::ITypedExpr>
+SubstraitVeloxExprConverter::toVeloxExpr(
+    const ::substrait::Expression::IfThen& ifThenExpr,
+    const RowTypePtr& inputType) {
+  VELOX_CHECK(ifThenExpr.ifs().size() > 0, "If clause expected.");
+
+  // Params are concatenated conditions and results with an optional "else" at
+  // the end, e.g. {condition1, result1, condition2, result2,..else}
+  std::vector<core::TypedExprPtr> params;
+  // If and then expressions are in pairs.
+  params.reserve(ifThenExpr.ifs().size() * 2);
+  std::optional<TypePtr> outputType = std::nullopt;
+  for (const auto& ifThen : ifThenExpr.ifs()) {
+    params.emplace_back(toVeloxExpr(ifThen.if_(), inputType));
+    const auto& thenExpr = toVeloxExpr(ifThen.then(), inputType);
+    // Get output type from the first then expression.
+    if (!outputType.has_value()) {
+      outputType = thenExpr->type();
+    }
+    params.emplace_back(thenExpr);
+  }
+
+  if (ifThenExpr.has_else_()) {
+    params.reserve(1);
+    params.emplace_back(toVeloxExpr(ifThenExpr.else_(), inputType));
+  }
+
+  VELOX_CHECK(outputType.has_value(), "Output type should be set.");
+  if (ifThenExpr.ifs().size() == 1) {
+    // If there is only one if-then clause, use if expression.
+    return std::make_shared<const core::CallTypedExpr>(
+        outputType.value(), std::move(params), "if");
+  }
+  return std::make_shared<const core::CallTypedExpr>(
+      outputType.value(), std::move(params), "switch");
+}
+
 std::shared_ptr<const core::ITypedExpr>
 SubstraitVeloxExprConverter::toVeloxExpr(
     const ::substrait::Expression_IfThen& substraitIfThen,
diff --git a/velox/substrait/SubstraitToVeloxExpr.h b/velox/substrait/SubstraitToVeloxExpr.h
index 8fd904d09dd3..dd7c3f6b8553 100644
--- a/velox/substrait/SubstraitToVeloxExpr.h
+++ b/velox/substrait/SubstraitToVeloxExpr.h
@@ -20,6 +20,11 @@
 #include "velox/substrait/SubstraitParser.h"
 #include "velox/vector/ComplexVector.h"
 
+#include "velox/type/StringView.h"
+#include "velox/vector/FlatVector.h"
+
+#include "velox/vector/ComplexVector.h"
+
 namespace facebook::velox::substrait {
 
 /// This class is used to convert Substrait representations to Velox
@@ -34,6 +39,12 @@ class SubstraitVeloxExprConverter {
       const std::unordered_map<uint64_t, std::string>& functionMap)
       : pool_(pool), functionMap_(functionMap) {}
 
+  /// Stores the variant and its type.
+  struct TypedVariant {
+    variant veloxVariant;
+    TypePtr variantType;
+  };
+
   /// Convert Substrait Field into Velox Field Expression.
   std::shared_ptr<const core::FieldAccessTypedExpr> toVeloxExpr(
       const ::substrait::Expression::FieldReference& substraitField,
@@ -49,7 +60,16 @@ class SubstraitVeloxExprConverter {
       const ::substrait::Expression::Cast& castExpr,
       const RowTypePtr& inputType);
 
-  /// Convert Substrait Literal into Velox Expression.
+  /// Create expression for alias.
+  std::shared_ptr<const core::ITypedExpr> toAliasExpr(
+      const std::vector<std::shared_ptr<const core::ITypedExpr>>& params);
+
+  /// Create expression for is_not_null.
+  std::shared_ptr<const core::ITypedExpr> toIsNotNullExpr(
+      const std::vector<std::shared_ptr<const core::ITypedExpr>>& params,
+      const TypePtr& outputType);
+
+  /// Used to convert Substrait Literal into Velox Expression.
   std::shared_ptr<const core::ConstantTypedExpr> toVeloxExpr(
       const ::substrait::Expression::Literal& substraitLit);
 
@@ -58,9 +78,9 @@ class SubstraitVeloxExprConverter {
       const ::substrait::Expression& substraitExpr,
       const RowTypePtr& inputType);
 
-  /// Convert Substrait IfThen into Velox Expression.
+  /// Convert Substrait IfThen into switch or if expression.
   std::shared_ptr<const core::ITypedExpr> toVeloxExpr(
-      const ::substrait::Expression::IfThen& substraitIfThen,
+      const ::substrait::Expression::IfThen& ifThenExpr,
       const RowTypePtr& inputType);
 
  private:
diff --git a/velox/substrait/SubstraitToVeloxPlan.cpp b/velox/substrait/SubstraitToVeloxPlan.cpp
index 746510371b12..55fe0a95d4e9 100644
--- a/velox/substrait/SubstraitToVeloxPlan.cpp
+++ b/velox/substrait/SubstraitToVeloxPlan.cpp
@@ -62,6 +62,31 @@ core::SortOrder toSortOrder(const ::substrait::SortField& sortField) {
 
 } // namespace
 
+template <typename T>
+// Get the lowest value for numeric type.
+T getLowest() {
+  return std::numeric_limits<T>::lowest();
+};
+
+// Get the lowest value for string.
+template <>
+std::string getLowest<std::string>() {
+  return "";
+};
+
+// Get the max value for numeric type.
+template <typename T>
+T getMax() {
+  return std::numeric_limits<T>::max();
+};
+
+// The max value will be used in BytesRange. Return empty string here instead.
+template <>
+std::string getMax<std::string>() {
+  return "";
+};
+} // namespace
+
 core::PlanNodePtr SubstraitVeloxPlanConverter::toVeloxPlan(
     const ::substrait::JoinRel& sJoin) {
   if (!sJoin.has_left()) {
@@ -166,13 +191,11 @@ std::shared_ptr<const core::PlanNode> SubstraitVeloxPlanConverter::toVeloxAgg(
       veloxGroupingExprs;
 
   // Get the grouping expressions.
-  uint32_t groupingOutIdx = 0;
   for (const auto& grouping : sAgg.groupings()) {
     for (const auto& groupingExpr : grouping.grouping_expressions()) {
       // Velox's groupings are limited to be Field.
       veloxGroupingExprs.emplace_back(
           exprConverter_->toVeloxExpr(groupingExpr.selection(), inputType));
-      groupingOutIdx += 1;
     }
   }
 
@@ -180,6 +203,7 @@ std::shared_ptr<const core::PlanNode> SubstraitVeloxPlanConverter::toVeloxAgg(
   // Each measure represents one aggregate expression.
   std::vector<std::shared_ptr<const core::CallTypedExpr>> aggExprs;
   aggExprs.reserve(sAgg.measures().size());
+
   for (const auto& smea : sAgg.measures()) {
     core::FieldAccessTypedExprPtr aggregateMask;
     ::substrait::Expression substraitAggMask = measure.filter();
@@ -219,7 +243,8 @@ std::shared_ptr<const core::PlanNode> SubstraitVeloxPlanConverter::toVeloxAgg(
   // Get the output names of Aggregation.
   std::vector<std::string> aggOutNames;
   aggOutNames.reserve(sAgg.measures().size());
-  for (int idx = groupingOutIdx; idx < groupingOutIdx + sAgg.measures().size();
+  for (int idx = veloxGroupingExprs.size();
+       idx < veloxGroupingExprs.size() + sAgg.measures().size();
        idx++) {
     aggOutNames.emplace_back(subParser_->makeNodeName(planNodeId_, idx));
   }
@@ -517,6 +542,7 @@ core::PlanNodePtr SubstraitVeloxPlanConverter::toVeloxPlan(
 
 core::PlanNodePtr SubstraitVeloxPlanConverter::toVeloxPlan(
     const ::substrait::ReadRel& readRel,
+    memory::MemoryPool* pool,
     const RowTypePtr& type) {
   ::substrait::ReadRel_VirtualTable readVirtualTable = readRel.virtual_table();
   int64_t numVectors = readVirtualTable.values_size();
@@ -559,11 +585,11 @@ core::PlanNodePtr SubstraitVeloxPlanConverter::toVeloxPlan(
         }
       }
       children.emplace_back(
-          setVectorFromVariants(outputChildType, batchChild, pool_));
+          setVectorFromVariants(outputChildType, batchChild, pool));
     }
 
     vectors.emplace_back(
-        std::make_shared<RowVector>(pool_, type, nullptr, batchSize, children));
+        std::make_shared<RowVector>(pool, type, nullptr, batchSize, children));
   }
   return std::make_shared<core::ValuesNode>(nextPlanNodeId(), vectors);
 }
@@ -626,7 +652,6 @@ core::PlanNodePtr SubstraitVeloxPlanConverter::toVeloxPlan(
   if (sRel.has_rel()) {
     return toVeloxPlan(sRel.rel());
   }
-
   VELOX_FAIL("RelRoot or Rel is expected in Plan.");
 }
 
@@ -705,21 +730,20 @@ class FilterInfo {
 connector::hive::SubfieldFilters SubstraitVeloxPlanConverter::toVeloxFilter(
     const std::vector<std::string>& inputNameList,
     const std::vector<TypePtr>& inputTypeList,
-    const ::substrait::Expression& substraitFilter) {
+    const ::substrait::Expression& sFilter) {
   connector::hive::SubfieldFilters filters;
   // A map between the column index and the FilterInfo for that column.
   std::unordered_map<int, std::shared_ptr<FilterInfo>> colInfoMap;
   for (int idx = 0; idx < inputNameList.size(); idx++) {
     colInfoMap[idx] = std::make_shared<FilterInfo>();
   }
-
   std::vector<::substrait::Expression_ScalarFunction> scalarFunctions;
-  flattenConditions(substraitFilter, scalarFunctions);
+  flattenConditions(sFilter, scalarFunctions);
   // Construct the FilterInfo for the related column.
   for (const auto& scalarFunction : scalarFunctions) {
-    auto filterNameSpec = substraitParser_->findFunctionSpec(
+    auto filterNameSpec = subParser_->findSubstraitFuncSpec(
         functionMap_, scalarFunction.function_reference());
-    auto filterName = substraitParser_->getFunctionName(filterNameSpec);
+    auto filterName = subParser_->getSubFunctionName(filterNameSpec);
     int32_t colIdx;
     // TODO: Add different types' support here.
     double val;
@@ -730,7 +754,7 @@ connector::hive::SubfieldFilters SubstraitVeloxPlanConverter::toVeloxFilter(
           auto sel = param.selection();
           // TODO: Only direct reference is considered here.
           auto dRef = sel.direct_reference();
-          colIdx = substraitParser_->parseReferenceSegment(dRef);
+          colIdx = subParser_->parseReferenceSegment(dRef);
           break;
         }
         case ::substrait::Expression::RexTypeCase::kLiteral: {
@@ -760,7 +784,6 @@ connector::hive::SubfieldFilters SubstraitVeloxPlanConverter::toVeloxFilter(
           filterName);
     }
   }
-
   // Construct the Filters.
   for (int idx = 0; idx < inputNameList.size(); idx++) {
     auto filterInfo = colInfoMap[idx];
@@ -826,67 +849,611 @@ std::string SubstraitVeloxPlanConverter::findFuncSpec(uint64_t id) {
   return subParser_->findSubstraitFuncSpec(functionMap_, id);
 }
 
-int32_t SubstraitVeloxPlanConverter::streamIsInput(
-    const ::substrait::ReadRel& sRead) {
-  if (sRead.has_local_files()) {
-    const auto& fileList = sRead.local_files().items();
-    if (fileList.size() == 0) {
-      VELOX_FAIL("At least one file path is expected.");
+core::PlanNodePtr SubstraitVeloxPlanConverter::toVeloxPlan(
+    const ::substrait::Plan& substraitPlan,
+    memory::MemoryPool* pool) {
+  VELOX_CHECK(
+      checkTypeExtension(substraitPlan),
+      "The type extension only have unknown type.")
+  // Construct the function map based on the Substrait representation.
+  constructFunctionMap(substraitPlan);
+
+  // In fact, only one RelRoot or Rel is expected here.
+  VELOX_CHECK_EQ(substraitPlan.relations_size(), 1);
+  const auto& rel = substraitPlan.relations(0);
+  if (rel.has_root()) {
+    return toVeloxPlan(rel.root(), pool);
+  }
+  if (rel.has_rel()) {
+    return toVeloxPlan(rel.rel(), pool);
+  }
+  VELOX_FAIL("Input is expected in RelRoot.");
+}
+
+core::PlanNodePtr SubstraitVeloxPlanConverter::toVeloxPlan(
+    const ::substrait::Plan& substraitPlan,
+    memory::MemoryPool* pool) {
+  // Construct the function map based on the Substrait representation.
+  constructFuncMap(sPlan);
+  VELOX_FAIL("RelRoot or Rel is expected in Plan.");
+}
+
+void SubstraitVeloxPlanConverter::extractJoinKeys(
+    const ::substrait::Expression& joinExpression,
+    std::vector<const ::substrait::Expression::FieldReference*>& leftExprs,
+    std::vector<const ::substrait::Expression::FieldReference*>& rightExprs) {
+  std::vector<const ::substrait::Expression*> expressions;
+  expressions.push_back(&joinExpression);
+  while (!expressions.empty()) {
+    auto visited = expressions.back();
+    expressions.pop_back();
+    if (visited->rex_type_case() ==
+        ::substrait::Expression::RexTypeCase::kScalarFunction) {
+      const auto& funcName =
+          subParser_->getSubFunctionName(subParser_->findVeloxFunction(
+              functionMap_, visited->scalar_function().function_reference()));
+      const auto& args = visited->scalar_function().args();
+      if (funcName == "and") {
+        expressions.push_back(&args[0]);
+        expressions.push_back(&args[1]);
+      } else if (funcName == "eq") {
+        VELOX_CHECK(std::all_of(
+            args.cbegin(), args.cend(), [](const ::substrait::Expression& arg) {
+              return arg.has_selection();
+            }));
+        leftExprs.push_back(&args[0].selection());
+        rightExprs.push_back(&args[1].selection());
+      } else {
+        VELOX_NYI("Join condition {} not supported.", funcName);
+      }
+    } else {
+      VELOX_FAIL(
+          "Unable to parse from join expression: {}",
+          joinExpression.DebugString());
+    }
+  }
+}
+
+connector::hive::SubfieldFilters SubstraitVeloxPlanConverter::toSubfieldFilters(
+    const std::vector<std::string>& inputNameList,
+    const std::vector<TypePtr>& inputTypeList,
+    const std::vector<::substrait::Expression_ScalarFunction>&
+        scalarFunctions) {
+  // A map between the column index and the FilterInfo.
+  std::unordered_map<uint32_t, std::shared_ptr<FilterInfo>> colInfoMap;
+  for (uint32_t idx = 0; idx < inputTypeList.size(); idx++) {
+    colInfoMap[idx] = std::make_shared<FilterInfo>();
+  }
+
+  // Construct the FilterInfo for the related column.
+  for (const auto& scalarFunction : scalarFunctions) {
+    auto filterNameSpec = subParser_->findSubstraitFuncSpec(
+        functionMap_, scalarFunction.function_reference());
+    auto filterName = subParser_->getSubFunctionName(filterNameSpec);
+    if (filterName == "not") {
+      VELOX_CHECK(scalarFunction.args().size() == 1);
+      VELOX_CHECK(
+          scalarFunction.args()[0].has_scalar_function(),
+          "Scalar function expected.");
+      // Set its chid to filter info with reverse enabled.
+      setFilterMap(
+          scalarFunction.args()[0].scalar_function(),
+          inputTypeList,
+          colInfoMap,
+          true);
+      continue;
+    }
+
+    if (filterName == "or") {
+      VELOX_CHECK(scalarFunction.args().size() == 2);
+      VELOX_CHECK(std::all_of(
+          scalarFunction.args().cbegin(),
+          scalarFunction.args().cend(),
+          [](const ::substrait::Expression& arg) {
+            return arg.has_scalar_function();
+          }));
+      // Set the chidren functions to filter info. They should be
+      // effective to the same field.
+      for (const auto& arg : scalarFunction.args()) {
+        setFilterMap(arg.scalar_function(), inputTypeList, colInfoMap);
+      }
+      continue;
+    }
+
+    setFilterMap(scalarFunction, inputTypeList, colInfoMap);
+  }
+
+  // Create subfield filters based on the constructed filter info map.
+  return mapToFilters(inputNameList, inputTypeList, colInfoMap);
+}
+
+bool SubstraitVeloxPlanConverter::fieldOrWithLiteral(
+    const ::substrait::Expression_ScalarFunction& function) {
+  if (function.args().size() == 1) {
+    if (function.args()[0].has_selection()) {
+      // Only field exists.
+      return true;
+    } else {
+      return false;
     }
+  }
 
-    // The stream input will be specified with the format of
-    // "iterator:${index}".
-    std::string filePath = fileList[0].uri_file();
-    std::string prefix = "iterator:";
-    std::size_t pos = filePath.find(prefix);
-    if (pos == std::string::npos) {
-      return -1;
+  if (function.args().size() != 2) {
+    return false;
+  }
+  bool fieldExists = false;
+  bool literalExists = false;
+  for (const auto& param : function.args()) {
+    auto typeCase = param.rex_type_case();
+    switch (typeCase) {
+      case ::substrait::Expression::RexTypeCase::kSelection:
+        fieldExists = true;
+        break;
+      case ::substrait::Expression::RexTypeCase::kLiteral:
+        literalExists = true;
+        break;
+      default:
+        break;
     }
+  }
+  // Whether the field and literal both exist.
+  return fieldExists && literalExists;
+}
 
-    // Get the index.
-    std::string idxStr = filePath.substr(pos + prefix.size(), filePath.size());
-    try {
-      return stoi(idxStr);
-    } catch (const std::exception& err) {
-      VELOX_FAIL(err.what());
+bool SubstraitVeloxPlanConverter::chidrenFunctionsOnSameField(
+    const ::substrait::Expression_ScalarFunction& function) {
+  // Get the column indices of the chidren functions.
+  std::vector<int32_t> colIndices;
+  for (const auto& arg : function.args()) {
+    if (!arg.has_scalar_function()) {
+      return false;
+    }
+    auto scalarFunction = arg.scalar_function();
+    for (const auto& param : scalarFunction.args()) {
+      if (param.has_selection()) {
+        auto field = param.selection();
+        VELOX_CHECK(field.has_direct_reference());
+        int32_t colIdx =
+            subParser_->parseReferenceSegment(field.direct_reference());
+        colIndices.emplace_back(colIdx);
+      }
     }
   }
-  if (validationMode_) {
-    return -1;
+
+  if (std::all_of(colIndices.begin(), colIndices.end(), [&](uint32_t idx) {
+        return idx == colIndices[0];
+      })) {
+    // All indices are the same.
+    return true;
   }
-  VELOX_FAIL("Local file is expected.");
+  return false;
 }
 
-void SubstraitVeloxPlanConverter::flattenConditions(
-    const ::substrait::Expression& substraitFilter,
-    std::vector<::substrait::Expression_ScalarFunction>& scalarFunctions,
-    std::vector<::substrait::Expression_SingularOrList>& singularOrLists) {
-  auto typeCase = substraitFilter.rex_type_case();
-  switch (typeCase) {
-    case ::substrait::Expression::RexTypeCase::kScalarFunction: {
-      auto sFunc = substraitFilter.scalar_function();
-      auto filterNameSpec = subParser_->findSubstraitFuncSpec(
-          functionMap_, sFunc.function_reference());
-      // TODO: Only and relation is supported here.
-      if (subParser_->getSubFunctionName(filterNameSpec) == "and") {
-        for (const auto& sCondition : sFunc.arguments()) {
-          flattenConditions(
-              getExprFromFunctionArgument(sCondition),
-              scalarFunctions,
-              singularOrLists);
-        }
+void SubstraitVeloxPlanConverter::separateFilters(
+    const std::vector<::substrait::Expression_ScalarFunction>& scalarFunctions,
+    std::vector<::substrait::Expression_ScalarFunction>& subfieldFunctions,
+    std::vector<::substrait::Expression_ScalarFunction>& remainingFunctions) {
+  // Condtions can be pushed down.
+  std::unordered_set<std::string> supportedFunctions = {
+      "is_not_null", "gte", "gt", "lte", "lt", "equal", "in"};
+  // Used to record the columns indices for not(equal) conditions.
+  std::unordered_set<uint32_t> notEqualCols;
+
+  for (const auto& scalarFunction : scalarFunctions) {
+    auto filterNameSpec = subParser_->findSubstraitFuncSpec(
+        functionMap_, scalarFunction.function_reference());
+    auto filterName = subParser_->getSubFunctionName(filterNameSpec);
+    if (filterName != "not" && filterName != "or") {
+      // Check if the condition is supported to be pushed down.
+      // The arg should be field or field with literal.
+      if (supportedFunctions.find(filterName) != supportedFunctions.end() &&
+          fieldOrWithLiteral(scalarFunction)) {
+        subfieldFunctions.emplace_back(scalarFunction);
       } else {
-        scalarFunctions.emplace_back(sFunc);
+        remainingFunctions.emplace_back(scalarFunction);
       }
+      continue;
+    }
+
+    // When the function is not/or, check whether its chidren can be
+    // pushed down. OR Conditon whose chidren functions are on different
+    // columns is not supported to be pushed down.
+    if (filterName == "or" && !chidrenFunctionsOnSameField(scalarFunction)) {
+      remainingFunctions.emplace_back(scalarFunction);
+      continue;
+    }
+  }
+
+  // pushed down. If yes, this scalar function will be added
+  // into the subfield functions.
+  bool supported = true;
+  for (const auto& arg : scalarFunction.args()) {
+    if (!arg.has_scalar_function()) {
+      // Not with a Boolean Literal is not supported curretly.
+      // It can be pushed down with an AlwaysTrue or AlwaysFalse Range.
+      supported = false;
       break;
     }
-    case ::substrait::Expression::RexTypeCase::kSingularOrList: {
-      singularOrLists.emplace_back(substraitFilter.singular_or_list());
+
+    auto nameSpec = subParser_->findSubstraitFuncSpec(
+        functionMap_, arg.scalar_function().function_reference());
+    auto functionName = subParser_->getSubFunctionName(nameSpec);
+
+    // The arg should be field or field with literal.
+    if (supportedFunctions.find(functionName) == supportedFunctions.end() ||
+        !fieldOrWithLiteral(arg.scalar_function())) {
+      supported = false;
       break;
     }
+
+    // Mutiple not(equal) conditons cannot be pushed down because
+    // the multiple range is in OR relation while AND relation is
+    // actually needed.
+    if (filterName == "not" && functionName == "equal") {
+      for (const auto& eqArg : arg.scalar_function().args()) {
+        if (!eqArg.has_selection()) {
+          continue;
+        }
+        uint32_t colIdx = subParser_->parseReferenceSegment(
+            eqArg.selection().direct_reference());
+        // If one not(equal) condition for this column already exists,
+        // this function cannot be pushed down then.
+        if (notEqualCols.find(colIdx) == notEqualCols.end()) {
+          notEqualCols.insert(colIdx);
+        } else {
+          supported = false;
+          break;
+        }
+      }
+      if (!supported) {
+        break;
+      }
+    }
+  }
+  if (supported) {
+    subfieldFunctions.emplace_back(scalarFunction);
+  } else {
+    remainingFunctions.emplace_back(scalarFunction);
+  }
+}
+}
+
+void SubstraitVeloxPlanConverter::setInValues(
+    const ::substrait::Expression_ScalarFunction& scalarFunction,
+    std::unordered_map<uint32_t, std::shared_ptr<FilterInfo>>& colInfoMap) {
+  VELOX_CHECK(
+      scalarFunction.args().size() == 2, "Two args expected in In expression.");
+  VELOX_CHECK(scalarFunction.args()[0].has_selection(), "Field expected.");
+
+  // Get the column index.
+  uint32_t colIdx = subParser_->parseReferenceSegment(
+      scalarFunction.args()[0].selection().direct_reference());
+  VELOX_CHECK(scalarFunction.args()[1].has_literal(), "Literal expected.");
+  VELOX_CHECK(scalarFunction.args()[1].literal().has_list(), "List expected.");
+
+  // Get the value list.
+  std::vector<variant> variants;
+  auto valueList = scalarFunction.args()[1].literal().list();
+  variants.reserve(valueList.values().size());
+  for (const auto& literal : valueList.values()) {
+    variants.emplace_back(
+        exprConverter_->toTypedVariant(literal)->veloxVariant);
+  }
+
+  // Set the value list to filter info.
+  colInfoMap[colIdx]->setValues(variants);
+}
+
+template <typename T>
+void SubstraitVeloxPlanConverter::setColInfoMap(
+    const std::string& filterName,
+    uint32_t colIdx,
+    std::optional<variant> literalVariant,
+    bool reverse,
+    std::unordered_map<uint32_t, std::shared_ptr<FilterInfo>>& colInfoMap) {
+  if (filterName == "is_not_null") {
+    if (reverse) {
+      VELOX_NYI("Reverse not supported for filter name '{}'", filterName);
+    }
+    colInfoMap[colIdx]->forbidsNull();
+    return;
+  }
+
+  if (filterName == "gte") {
+    if (reverse) {
+      colInfoMap[colIdx]->setUpper(literalVariant, true);
+    } else {
+      colInfoMap[colIdx]->setLower(literalVariant, false);
+    }
+    return;
+  }
+
+  if (filterName == "gt") {
+    if (reverse) {
+      colInfoMap[colIdx]->setUpper(literalVariant, false);
+    } else {
+      colInfoMap[colIdx]->setLower(literalVariant, true);
+    }
+    return;
+  }
+
+  if (filterName == "lte") {
+    if (reverse) {
+      colInfoMap[colIdx]->setLower(literalVariant, true);
+    } else {
+      colInfoMap[colIdx]->setUpper(literalVariant, false);
+    }
+    return;
+  }
+
+  if (filterName == "lt") {
+    if (reverse) {
+      colInfoMap[colIdx]->setLower(literalVariant, false);
+    } else {
+      colInfoMap[colIdx]->setUpper(literalVariant, true);
+    }
+    return;
+  }
+
+  if (filterName == "equal") {
+    if (reverse) {
+      colInfoMap[colIdx]->setNotValue(literalVariant);
+    } else {
+      colInfoMap[colIdx]->setLower(literalVariant, false);
+      colInfoMap[colIdx]->setUpper(literalVariant, false);
+    }
+    return;
+  }
+  VELOX_NYI("SetColInfoMap not supported for filter name '{}'", filterName);
+}
+
+void SubstraitVeloxPlanConverter::setFilterMap(
+    const ::substrait::Expression_ScalarFunction& scalarFunction,
+    const std::vector<TypePtr>& inputTypeList,
+    std::unordered_map<uint32_t, std::shared_ptr<FilterInfo>>& colInfoMap,
+    bool reverse) {
+  auto nameSpec = subParser_->findSubstraitFuncSpec(
+      functionMap_, scalarFunction.function_reference());
+  auto functionName = subParser_->getSubFunctionName(nameSpec);
+
+  // Extract the column index and column bound from the scalar function.
+  std::optional<uint32_t> colIdx;
+  std::optional<::substrait::Expression_Literal> substraitLit;
+  for (const auto& param : scalarFunction.arguments()) {
+    auto typeCase = getExprFromFunctionArgument(param).rex_type_case();
+    switch (typeCase) {
+      case ::substrait::Expression::RexTypeCase::kSelection:
+        colIdx = subParser_->parseReferenceSegment(
+            getExprFromFunctionArgument(param).selection().direct_reference());
+        break;
+      case ::substrait::Expression::RexTypeCase::kLiteral:
+        substraitLit = getExprFromFunctionArgument(param).literal();
+        break;
+      default:
+        VELOX_NYI(
+            "Substrait conversion not supported for arg type '{}'", typeCase);
+    }
+  }
+  if (!colIdx.has_value()) {
+    VELOX_NYI("Column index is expected in subfield filters creation.");
+  }
+
+  // Set the extracted bound to the specific column.
+  uint32_t colIdxVal = colIdx.value();
+  auto inputType = inputTypeList[colIdxVal];
+  std::optional<variant> val;
+  switch (inputType->kind()) {
+    case TypeKind::INTEGER:
+      if (substraitLit) {
+        val = variant(substraitLit.value().i32());
+      }
+      setColInfoMap<int>(functionName, colIdxVal, val, reverse, colInfoMap);
+      break;
+    case TypeKind::BIGINT:
+      if (substraitLit) {
+        val = variant(substraitLit.value().i64());
+      }
+      setColInfoMap<int64_t>(functionName, colIdxVal, val, reverse, colInfoMap);
+      break;
+    case TypeKind::DOUBLE:
+      if (substraitLit) {
+        val = variant(substraitLit.value().fp64());
+      }
+      setColInfoMap<double>(functionName, colIdxVal, val, reverse, colInfoMap);
+      break;
+    case TypeKind::VARCHAR:
+      if (substraitLit) {
+        val = variant(substraitLit.value().string());
+      }
+      setColInfoMap<std::string>(
+          functionName, colIdxVal, val, reverse, colInfoMap);
+      break;
     default:
-      VELOX_NYI("GetFlatConditions not supported for type '{}'", typeCase);
+      VELOX_NYI(
+          "Subfield filters creation not supported for input type '{}'",
+          inputType);
+  }
+}
+
+template <TypeKind KIND, typename FilterType>
+void SubstraitVeloxPlanConverter::createNotEqualFilter(
+    variant notVariant,
+    bool nullAllowed,
+    std::vector<std::unique_ptr<FilterType>>& colFilters) {
+  using T = typename RangeTraits<KIND>::NativeType;
+  using RangeType = typename RangeTraits<KIND>::RangeType;
+
+  // Value > lower
+  std::unique_ptr<FilterType> lowerFilter = std::make_unique<RangeType>(
+      notVariant.value<T>(), /*lower*/
+      false, /*lowerUnbounded*/
+      true, /*lowerExclusive*/
+      getMax<T>(), /*upper*/
+      true, /*upperUnbounded*/
+      false, /*upperExclusive*/
+      nullAllowed); /*nullAllowed*/
+  colFilters.emplace_back(std::move(lowerFilter));
+
+  // Value < upper
+  std::unique_ptr<FilterType> upperFilter = std::make_unique<RangeType>(
+      getLowest<T>(), /*lower*/
+      true, /*lowerUnbounded*/
+      false, /*lowerExclusive*/
+      notVariant.value<T>(), /*upper*/
+      false, /*upperUnbounded*/
+      true, /*upperExclusive*/
+      nullAllowed); /*nullAllowed*/
+  colFilters.emplace_back(std::move(upperFilter));
+}
+
+template <TypeKind KIND>
+void SubstraitVeloxPlanConverter::setInFilter(
+    const std::vector<variant>& variants,
+    bool nullAllowed,
+    const std::string& inputName,
+    connector::hive::SubfieldFilters& filters) {}
+
+template <>
+void SubstraitVeloxPlanConverter::setInFilter<TypeKind::DOUBLE>(
+    const std::vector<variant>& variants,
+    bool nullAllowed,
+    const std::string& inputName,
+    connector::hive::SubfieldFilters& filters) {
+  std::vector<double> values;
+  values.reserve(variants.size());
+  for (const auto& variant : variants) {
+    double value = variant.value<double>();
+    values.emplace_back(value);
+  }
+  filters[common::Subfield(inputName)] =
+      common::createDoubleValues(values, nullAllowed);
+}
+
+template <>
+void SubstraitVeloxPlanConverter::setInFilter<TypeKind::BIGINT>(
+    const std::vector<variant>& variants,
+    bool nullAllowed,
+    const std::string& inputName,
+    connector::hive::SubfieldFilters& filters) {
+  std::vector<int64_t> values;
+  values.reserve(variants.size());
+  for (const auto& variant : variants) {
+    int64_t value = variant.value<int64_t>();
+    values.emplace_back(value);
+  }
+  filters[common::Subfield(inputName)] =
+      common::createBigintValues(values, nullAllowed);
+}
+
+template <>
+void SubstraitVeloxPlanConverter::setInFilter<TypeKind::VARCHAR>(
+    const std::vector<variant>& variants,
+    bool nullAllowed,
+    const std::string& inputName,
+    connector::hive::SubfieldFilters& filters) {
+  std::vector<std::string> values;
+  values.reserve(variants.size());
+  for (const auto& variant : variants) {
+    std::string value = variant.value<std::string>();
+    values.emplace_back(value);
+  }
+  filters[common::Subfield(inputName)] =
+      std::make_unique<common::BytesValues>(values, nullAllowed);
+}
+
+template <TypeKind KIND, typename FilterType>
+void SubstraitVeloxPlanConverter::setSubfieldFilter(
+    std::vector<std::unique_ptr<FilterType>> colFilters,
+    const std::string& inputName,
+    bool nullAllowed,
+    connector::hive::SubfieldFilters& filters) {
+  using MultiRangeType = typename RangeTraits<KIND>::MultiRangeType;
+
+  if (colFilters.size() == 1) {
+    filters[common::Subfield(inputName)] = std::move(colFilters[0]);
+  } else if (colFilters.size() > 1) {
+    filters[common::Subfield(inputName)] =
+        std::make_unique<MultiRangeType>(std::move(colFilters), nullAllowed);
+  }
+}
+
+template <TypeKind KIND, typename FilterType>
+void SubstraitVeloxPlanConverter::constructSubfieldFilters(
+    uint32_t colIdx,
+    const std::string& inputName,
+    const std::shared_ptr<FilterInfo>& filterInfo,
+    connector::hive::SubfieldFilters& filters) {
+  using NativeType = typename RangeTraits<KIND>::NativeType;
+  using RangeType = typename RangeTraits<KIND>::RangeType;
+  using MultiRangeType = typename RangeTraits<KIND>::MultiRangeType;
+
+  // Handle 'in' filter.
+  if (filterInfo->valuesVector_.size() > 0) {
+    setInFilter<KIND>(
+        filterInfo->valuesVector_,
+        filterInfo->nullAllowed_,
+        inputName,
+        filters);
+    // Currently, In cannot coexist with other filter conditions
+    // due to multirange is in 'OR' relation but 'AND' can be needed.
+    VELOX_CHECK(
+        filterInfo->lowerBounds_.size() == 0,
+        "Other conditons cannot be supported.");
+    VELOX_CHECK(
+        filterInfo->upperBounds_.size() == 0,
+        "Other conditons cannot be supported.");
+    return;
+  }
+
+  // Construct the Filters.
+  std::vector<std::unique_ptr<FilterType>> colFilters;
+  NativeType lowerBound = getLowest<NativeType>();
+  NativeType upperBound = getMax<NativeType>();
+  bool lowerUnbounded = true;
+  bool upperUnbounded = true;
+  bool lowerExclusive = false;
+  bool upperExclusive = false;
+  if (filterInfo->isInitialized()) {
+    // Handle not(equal) filter.
+    if (filterInfo->notValue_) {
+      variant notVariant = filterInfo->notValue_.value();
+      createNotEqualFilter<KIND, FilterType>(
+          notVariant, filterInfo->nullAllowed_, colFilters);
+    }
+
+    // Handle other filter ranges.
+    uint32_t rangeSize = std::max(
+        filterInfo->lowerBounds_.size(), filterInfo->upperBounds_.size());
+    bool nullAllowed = filterInfo->nullAllowed_;
+    for (uint32_t idx = 0; idx < rangeSize; idx++) {
+      if (idx < filterInfo->lowerBounds_.size() &&
+          filterInfo->lowerBounds_[idx]) {
+        lowerUnbounded = false;
+        variant lowerVariant = filterInfo->lowerBounds_[idx].value();
+        lowerBound = lowerVariant.value<NativeType>();
+        lowerExclusive = filterInfo->lowerExclusives_[idx];
+      }
+      if (idx < filterInfo->upperBounds_.size() &&
+          filterInfo->upperBounds_[idx]) {
+        upperUnbounded = false;
+        variant upperVariant = filterInfo->upperBounds_[idx].value();
+        upperBound = upperVariant.value<NativeType>();
+        upperExclusive = filterInfo->upperExclusives_[idx];
+      }
+      std::unique_ptr<FilterType> filter = std::make_unique<RangeType>(
+          lowerBound,
+          lowerUnbounded,
+          lowerExclusive,
+          upperBound,
+          upperUnbounded,
+          upperExclusive,
+          nullAllowed);
+      colFilters.emplace_back(std::move(filter));
+    }
   }
+  // Set the SubfieldFilter.
+  setSubfieldFilter<KIND, FilterType>(
+      std::move(colFilters), inputName, filterInfo->nullAllowed_, filters);
 }
 
 void SubstraitVeloxPlanConverter::constructFunctionMap(
@@ -901,6 +1468,8 @@ void SubstraitVeloxPlanConverter::constructFunctionMap(
     auto name = sFmap.name();
     functionMap_[id] = name;
   }
+  exprConverter_ =
+      std::make_shared<SubstraitVeloxExprConverter>(pool_, functionMap_);
 }
 
 bool SubstraitVeloxPlanConverter::checkTypeExtension(
@@ -937,17 +1506,21 @@ void SubstraitVeloxPlanConverter::extractJoinKeys(
       const auto& funcName =
           subParser_->getSubFunctionName(subParser_->findVeloxFunction(
               functionMap_, visited->scalar_function().function_reference()));
-      const auto& args = visited->scalar_function().args();
+      const auto& args = visited->scalar_function().arguments();
       if (funcName == "and") {
-        expressions.push_back(&args[0]);
-        expressions.push_back(&args[1]);
-      } else if (funcName == "equal") {
+        expressions.push_back(&getExprFromFunctionArgument(args[0]));
+        expressions.push_back(&getExprFromFunctionArgument(args[1]));
+      } else if (funcName == "eq") {
         VELOX_CHECK(std::all_of(
-            args.cbegin(), args.cend(), [](const ::substrait::Expression& arg) {
-              return arg.has_selection();
+            args.cbegin(),
+            args.cend(),
+            [](const ::substrait::FunctionArgument& arg) {
+              return getExprFromFunctionArgument(arg).has_selection();
             }));
-        leftExprs.push_back(&args[0].selection());
-        rightExprs.push_back(&args[1].selection());
+        leftExprs.push_back(&getExprFromFunctionArgument(args[0]).selection());
+        rightExprs.push_back(&getExprFromFunctionArgument(args[1]).selection());
+      } else {
+        VELOX_NYI("Join condition {} not supported.", funcName);
       }
     } else {
       VELOX_FAIL(
@@ -957,4 +1530,68 @@ void SubstraitVeloxPlanConverter::extractJoinKeys(
   }
 }
 
+connector::hive::SubfieldFilters SubstraitVeloxPlanConverter::mapToFilters(
+    const std::vector<std::string>& inputNameList,
+    const std::vector<TypePtr>& inputTypeList,
+    std::unordered_map<uint32_t, std::shared_ptr<FilterInfo>> colInfoMap) {
+  // Construct the subfield filters based on the filter info map.
+  connector::hive::SubfieldFilters filters;
+  for (uint32_t colIdx = 0; colIdx < inputNameList.size(); colIdx++) {
+    auto inputType = inputTypeList[colIdx];
+    switch (inputType->kind()) {
+      case TypeKind::INTEGER:
+        constructSubfieldFilters<TypeKind::BIGINT, common::BigintRange>(
+            colIdx, inputNameList[colIdx], colInfoMap[colIdx], filters);
+        break;
+      case TypeKind::BIGINT:
+        constructSubfieldFilters<TypeKind::BIGINT, common::BigintRange>(
+            colIdx, inputNameList[colIdx], colInfoMap[colIdx], filters);
+        break;
+      case TypeKind::DOUBLE:
+        constructSubfieldFilters<TypeKind::DOUBLE, common::Filter>(
+            colIdx, inputNameList[colIdx], colInfoMap[colIdx], filters);
+        break;
+      case TypeKind::VARCHAR:
+        constructSubfieldFilters<TypeKind::VARCHAR, common::Filter>(
+            colIdx, inputNameList[colIdx], colInfoMap[colIdx], filters);
+        break;
+      default:
+        VELOX_NYI(
+            "Subfield filters creation not supported for input type '{}'",
+            inputType);
+    }
+  }
+  return filters;
+}
+
+std::shared_ptr<const core::ITypedExpr>
+SubstraitVeloxPlanConverter::connectWithAnd(
+    std::vector<std::string> inputNameList,
+    std::vector<TypePtr> inputTypeList,
+    const std::vector<::substrait::Expression_ScalarFunction>&
+        remainingFunctions) {
+  if (remainingFunctions.size() == 0) {
+    return nullptr;
+  }
+  auto inputType = ROW(std::move(inputNameList), std::move(inputTypeList));
+  std::shared_ptr<const core::ITypedExpr> remainingFilter =
+      exprConverter_->toVeloxExpr(remainingFunctions[0], inputType);
+  if (remainingFunctions.size() == 1) {
+    return remainingFilter;
+  }
+  // Will connect multiple functions with AND.
+  uint32_t idx = 1;
+  while (idx < remainingFunctions.size()) {
+    std::vector<std::shared_ptr<const core::ITypedExpr>> params;
+    params.reserve(2);
+    params.emplace_back(remainingFilter);
+    params.emplace_back(
+        exprConverter_->toVeloxExpr(remainingFunctions[idx], inputType));
+    remainingFilter = std::make_shared<const core::CallTypedExpr>(
+        BOOLEAN(), std::move(params), "and");
+    idx += 1;
+  }
+  return remainingFilter;
+}
+
 } // namespace facebook::velox::substrait
diff --git a/velox/substrait/SubstraitToVeloxPlan.h b/velox/substrait/SubstraitToVeloxPlan.h
index 2b229fee9448..6a9052783b01 100644
--- a/velox/substrait/SubstraitToVeloxPlan.h
+++ b/velox/substrait/SubstraitToVeloxPlan.h
@@ -19,30 +19,36 @@
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/core/PlanNode.h"
 #include "velox/substrait/SubstraitToVeloxExpr.h"
+#include "velox/substrait/TypeUtils.h"
 
 namespace facebook::velox::substrait {
+struct SplitInfo {
+  /// Whether the split comes from arrow array stream node.
+  bool isStream = false;
 
-/// This class is used to convert the Substrait plan into Velox plan.
-class SubstraitVeloxPlanConverter {
- public:
-  explicit SubstraitVeloxPlanConverter(memory::MemoryPool* pool)
-      : pool_(pool) {}
-  struct SplitInfo {
-    /// The Partition index.
-    u_int32_t partitionIndex;
+  /// The Partition index.
+  u_int32_t partitionIndex;
 
-    /// The file paths to be scanned.
-    std::vector<std::string> paths;
+  /// The file paths to be scanned.
+  std::vector<std::string> paths;
 
-    /// The file starts in the scan.
-    std::vector<u_int64_t> starts;
+  /// The file starts in the scan.
+  std::vector<u_int64_t> starts;
 
-    /// The lengths to be scanned.
-    std::vector<u_int64_t> lengths;
+  /// The lengths to be scanned.
+  std::vector<u_int64_t> lengths;
 
-    /// The file format of the files to be scanned.
-    dwio::common::FileFormat format;
-  };
+  /// The file format of the files to be scanned.
+  dwio::common::FileFormat format;
+};
+
+/// This class is used to convert the Substrait plan into Velox plan.
+class SubstraitVeloxPlanConverter {
+ public:
+  SubstraitVeloxPlanConverter(
+      memory::MemoryPool* pool,
+      bool validationMode = false)
+      : pool_(pool), validationMode_(validationMode) {}
 
   /// Used to convert Substrait JoinRel into Velox PlanNode.
   core::PlanNodePtr toVeloxPlan(const ::substrait::JoinRel& sJoin);
@@ -173,6 +179,224 @@ class SubstraitVeloxPlanConverter {
       std::vector<const ::substrait::Expression::FieldReference*>& rightExprs);
 
  private:
+  /// Filter info for a column used in filter push down.
+  class FilterInfo {
+   public:
+    // Disable null allow.
+    void forbidsNull() {
+      nullAllowed_ = false;
+      if (!isInitialized_) {
+        isInitialized_ = true;
+      }
+    }
+
+    // Return the initialization status.
+    bool isInitialized() {
+      return isInitialized_ ? true : false;
+    }
+
+    // Add a lower bound to the range. Multiple lower bounds are
+    // regarded to be in 'or' relation.
+    void setLower(const std::optional<variant>& left, bool isExclusive) {
+      lowerBounds_.emplace_back(left);
+      lowerExclusives_.emplace_back(isExclusive);
+      if (!isInitialized_) {
+        isInitialized_ = true;
+      }
+    }
+
+    // Add a upper bound to the range. Multiple upper bounds are
+    // regarded to be in 'or' relation.
+    void setUpper(const std::optional<variant>& right, bool isExclusive) {
+      upperBounds_.emplace_back(right);
+      upperExclusives_.emplace_back(isExclusive);
+      if (!isInitialized_) {
+        isInitialized_ = true;
+      }
+    }
+
+    // Set a list of values to be used in the push down of 'in' expression.
+    void setValues(const std::vector<variant>& values) {
+      for (const auto& value : values) {
+        valuesVector_.emplace_back(value);
+      }
+      if (!isInitialized_) {
+        isInitialized_ = true;
+      }
+    }
+
+    // Set a value for the not(equal) condition.
+    void setNotValue(const std::optional<variant>& notValue) {
+      notValue_ = notValue;
+      if (!isInitialized_) {
+        isInitialized_ = true;
+      }
+    }
+
+    // Whether this filter map is initialized.
+    bool isInitialized_ = false;
+
+    // The null allow.
+    bool nullAllowed_ = true;
+
+    // If true, left bound will be exclusive.
+    std::vector<bool> lowerExclusives_;
+
+    // If true, right bound will be exclusive.
+    std::vector<bool> upperExclusives_;
+
+    // A value should not be equal to.
+    std::optional<variant> notValue_ = std::nullopt;
+
+    // The lower bounds in 'or' relation.
+    std::vector<std::optional<variant>> lowerBounds_;
+
+    // The upper bounds in 'or' relation.
+    std::vector<std::optional<variant>> upperBounds_;
+
+    // The list of values used in 'in' expression.
+    std::vector<variant> valuesVector_;
+  };
+
+  /// Helper Function to convert Substrait sortField to Velox sortingKeys and
+  /// sortingOrders.
+  std::pair<
+      std::vector<core::FieldAccessTypedExprPtr>,
+      std::vector<core::SortOrder>>
+  processSortField(
+      const ::google::protobuf::RepeatedPtrField<::substrait::SortField>&
+          sortField,
+      const RowTypePtr& inputType);
+
+  /// A function returning current function id and adding the plan node id by
+  /// one once called.
+  std::string nextPlanNodeId();
+
+  /// Check the args of the scalar function. Should be field or
+  /// field with literal.
+  bool fieldOrWithLiteral(
+      const ::substrait::Expression_ScalarFunction& function);
+
+  /// Separate the functions to be two parts:
+  /// subfield functions to be handled by the subfieldFilters in
+  /// HiveConnector, and remaining functions to be handled by the
+  /// remainingFilter in HiveConnector.
+  void separateFilters(
+      const std::vector<::substrait::Expression_ScalarFunction>&
+          scalarFunctions,
+      std::vector<::substrait::Expression_ScalarFunction>& subfieldFunctions,
+      std::vector<::substrait::Expression_ScalarFunction>& remainingFunctions);
+
+  /// Check whether the chidren functions of this scalar function have the
+  /// same column index. Curretly used to check whether the two chilren
+  /// functions of 'or' expression are effective on the same column.
+  bool chidrenFunctionsOnSameField(
+      const ::substrait::Expression_ScalarFunction& function);
+
+  /// Extract the list from in function, and set it to the filter info.
+  void setInValues(
+      const ::substrait::Expression_ScalarFunction& scalarFunction,
+      std::unordered_map<uint32_t, std::shared_ptr<FilterInfo>>& colInfoMap);
+
+  /// Extract the scalar function, and set the filter info for different types
+  /// of columns. If reverse is true, the opposite filter info will be set.
+  void setFilterMap(
+      const ::substrait::Expression_ScalarFunction& scalarFunction,
+      const std::vector<TypePtr>& inputTypeList,
+      std::unordered_map<uint32_t, std::shared_ptr<FilterInfo>>& colInfoMap,
+      bool reverse = false);
+
+  /// Set the filter info for a column base on the information
+  /// extracted from filter condition.
+  template <typename T>
+  void setColInfoMap(
+      const std::string& filterName,
+      uint32_t colIdx,
+      std::optional<variant> literalVariant,
+      bool reverse,
+      std::unordered_map<uint32_t, std::shared_ptr<FilterInfo>>& colInfoMap);
+
+  /// Create a multirange to specify the filter 'x != notValue' with:
+  /// x > notValue or x < notValue.
+  template <TypeKind KIND, typename FilterType>
+  void createNotEqualFilter(
+      variant notVariant,
+      bool nullAllowed,
+      std::vector<std::unique_ptr<FilterType>>& colFilters);
+
+  /// Create a values range to handle in filter.
+  /// variants: the list of values extracted from the in expression.
+  /// inputName: the column input name.
+  template <TypeKind KIND>
+  void setInFilter(
+      const std::vector<variant>& variants,
+      bool nullAllowed,
+      const std::string& inputName,
+      connector::hive::SubfieldFilters& filters);
+
+  /// Set the constructed filters into SubfieldFilters.
+  /// The FilterType is used to distinguish BigintRange and
+  /// Filter (the base class). This is needed because BigintMultiRange
+  /// can only accept the unique ptr of BigintRange as parameter.
+  template <TypeKind KIND, typename FilterType>
+  void setSubfieldFilter(
+      std::vector<std::unique_ptr<FilterType>> colFilters,
+      const std::string& inputName,
+      bool nullAllowed,
+      connector::hive::SubfieldFilters& filters);
+
+  /// Create the subfield filter based on the constructed filter info.
+  /// inputName: the input name of a column.
+  template <TypeKind KIND, typename FilterType>
+  void constructSubfieldFilters(
+      uint32_t colIdx,
+      const std::string& inputName,
+      const std::shared_ptr<FilterInfo>& filterInfo,
+      connector::hive::SubfieldFilters& filters);
+
+  /// Construct subfield filters according to the pre-set map of filter info.
+  connector::hive::SubfieldFilters mapToFilters(
+      const std::vector<std::string>& inputNameList,
+      const std::vector<TypePtr>& inputTypeList,
+      std::unordered_map<uint32_t, std::shared_ptr<FilterInfo>> colInfoMap);
+
+  /// Convert subfield functions into subfieldFilters to
+  /// be used in Hive Connector.
+  connector::hive::SubfieldFilters toSubfieldFilters(
+      const std::vector<std::string>& inputNameList,
+      const std::vector<TypePtr>& inputTypeList,
+      const std::vector<::substrait::Expression_ScalarFunction>&
+          subfieldFunctions);
+
+  /// Connect all remaining functions with 'and' relation
+  /// for the use of remaingFilter in Hive Connector.
+  std::shared_ptr<const core::ITypedExpr> connectWithAnd(
+      std::vector<std::string> inputNameList,
+      std::vector<TypePtr> inputTypeList,
+      const std::vector<::substrait::Expression_ScalarFunction>&
+          remainingFunctions);
+
+  /// Used to convert Substrait Filter into Velox SubfieldFilters which will
+  /// be used in TableScan.
+  connector::hive::SubfieldFilters toVeloxFilter(
+      const std::vector<std::string>& inputNameList,
+      const std::vector<TypePtr>& inputTypeList,
+      const ::substrait::Expression& substraitFilter);
+
+  /// Used to convert AggregateRel into Velox plan node.
+  /// The output of child node will be used as the input of Aggregation.
+  std::shared_ptr<const core::PlanNode> toVeloxAgg(
+      const ::substrait::AggregateRel& sAgg,
+      const std::shared_ptr<const core::PlanNode>& childNode,
+      const core::AggregationNode::Step& aggStep);
+
+  /// Helper function to convert the input of Substrait Rel to Velox Node.
+  template <typename T>
+  core::PlanNodePtr convertSingleInput(T rel) {
+    VELOX_CHECK(rel.has_input(), "Child Rel is expected here.");
+    return toVeloxPlan(rel.input());
+  }
+
   /// The Partition index.
   u_int32_t partitionIndex_;
 
@@ -202,32 +426,6 @@ class SubstraitVeloxPlanConverter {
   /// recognizable representations.
   std::shared_ptr<SubstraitParser> subParser_{
       std::make_shared<SubstraitParser>()};
-
-  /// Helper Function to convert Substrait sortField to Velox sortingKeys and
-  /// sortingOrders.
-  std::pair<
-      std::vector<core::FieldAccessTypedExprPtr>,
-      std::vector<core::SortOrder>>
-  processSortField(
-      const ::google::protobuf::RepeatedPtrField<::substrait::SortField>&
-          sortField,
-      const RowTypePtr& inputType);
-
-  /// The Expression converter used to convert Substrait representations into
-  /// Velox expressions.
-  std::shared_ptr<SubstraitVeloxExprConverter> exprConverter_;
-
-  /// A function returning current function id and adding the plan node id by
-  /// one once called.
-  std::string nextPlanNodeId();
-
-  /// Used to convert Substrait Filter into Velox SubfieldFilters which will
-  /// be used in TableScan.
-  connector::hive::SubfieldFilters toVeloxFilter(
-      const std::vector<std::string>& inputNameList,
-      const std::vector<TypePtr>& inputTypeList,
-      const ::substrait::Expression& substraitFilter);
-
   /// Mapping from leaf plan node ID to splits.
   std::unordered_map<core::PlanNodeId, std::shared_ptr<SplitInfo>>
       splitInfoMap_;
@@ -235,19 +433,12 @@ class SubstraitVeloxPlanConverter {
   /// Memory pool.
   memory::MemoryPool* pool_;
 
-  /// Helper function to convert the input of Substrait Rel to Velox Node.
-  template <typename T>
-  core::PlanNodePtr convertSingleInput(T rel) {
-    VELOX_CHECK(rel.has_input(), "Child Rel is expected here.");
-    return toVeloxPlan(rel.input());
-  }
+  /// A flag used to specify validation.
+  bool validationMode_ = false;
 
-  /// Used to convert AggregateRel into Velox plan node.
-  /// The output of child node will be used as the input of Aggregation.
-  std::shared_ptr<const core::PlanNode> toVeloxAgg(
-      const ::substrait::AggregateRel& sAgg,
-      const std::shared_ptr<const core::PlanNode>& childNode,
-      const core::AggregationNode::Step& aggStep);
+  /// The Expression converter used to convert Substrait representations into
+  /// Velox expressions.
+  std::shared_ptr<SubstraitVeloxExprConverter> exprConverter_;
 };
 
 } // namespace facebook::velox::substrait
diff --git a/velox/substrait/SubstraitToVeloxPlanValidator.cpp b/velox/substrait/SubstraitToVeloxPlanValidator.cpp
index cc412523fa95..eb9ee264aa6f 100644
--- a/velox/substrait/SubstraitToVeloxPlanValidator.cpp
+++ b/velox/substrait/SubstraitToVeloxPlanValidator.cpp
@@ -16,6 +16,7 @@
 
 #include "velox/substrait/SubstraitToVeloxPlanValidator.h"
 #include "TypeUtils.h"
+#include "velox/functions/sparksql/Register.h"
 
 #include "velox/functions/prestosql/registration/RegistrationFunctions.h"
 
@@ -78,7 +79,8 @@ bool SubstraitToVeloxPlanValidator::validate(
   const auto& extension = sProject.advanced_extension();
   std::vector<TypePtr> types;
   if (!validateInputTypes(extension, types)) {
-    std::cout << "Validation failed for input types in ProjectRel" << std::endl;
+    std::cout << "Validation failed for input types in ProjectRel."
+              << std::endl;
     return false;
   }
 
@@ -101,7 +103,7 @@ bool SubstraitToVeloxPlanValidator::validate(
     }
     // Try to compile the expressions. If there is any unregistred funciton or
     // mismatched type, exception will be thrown.
-    exec::ExprSet exprSet(std::move(expressions), &execCtx_);
+    exec::ExprSet exprSet(std::move(expressions), execCtx_);
   } catch (const VeloxException& err) {
     std::cout << "Validation failed for expression in ProjectRel due to:"
               << err.message() << std::endl;
@@ -112,7 +114,45 @@ bool SubstraitToVeloxPlanValidator::validate(
 
 bool SubstraitToVeloxPlanValidator::validate(
     const ::substrait::FilterRel& sFilter) {
-  return false;
+  if (sFilter.has_input() && !validate(sFilter.input())) {
+    return false;
+  }
+
+  // Get and validate the input types from extension.
+  if (!sFilter.has_advanced_extension()) {
+    std::cout << "Input types are expected in FilterRel." << std::endl;
+    return false;
+  }
+  const auto& extension = sFilter.advanced_extension();
+  std::vector<TypePtr> types;
+  if (!validateInputTypes(extension, types)) {
+    std::cout << "Validation failed for input types in FilterRel." << std::endl;
+    return false;
+  }
+
+  int32_t inputPlanNodeId = 0;
+  // Create the fake input names to be used in row type.
+  std::vector<std::string> names;
+  names.reserve(types.size());
+  for (uint32_t colIdx = 0; colIdx < types.size(); colIdx++) {
+    names.emplace_back(subParser_->makeNodeName(inputPlanNodeId, colIdx));
+  }
+  auto rowType = std::make_shared<RowType>(std::move(names), std::move(types));
+
+  std::vector<std::shared_ptr<const core::ITypedExpr>> expressions;
+  expressions.reserve(1);
+  try {
+    expressions.emplace_back(
+        exprConverter_->toVeloxExpr(sFilter.condition(), rowType));
+    // Try to compile the expressions. If there is any unregistred funciton
+    // or mismatched type, exception will be thrown.
+    exec::ExprSet exprSet(std::move(expressions), execCtx_);
+  } catch (const VeloxException& err) {
+    std::cout << "Validation failed for expression in ProjectRel due to:"
+              << err.message() << std::endl;
+    return false;
+  }
+  return true;
 }
 
 bool SubstraitToVeloxPlanValidator::validate(
@@ -174,7 +214,7 @@ bool SubstraitToVeloxPlanValidator::validate(
     try {
       auto expression =
           exprConverter_->toVeloxExpr(sJoin.post_join_filter(), rowType);
-      exec::ExprSet exprSet({std::move(expression)}, &execCtx_);
+      exec::ExprSet exprSet({std::move(expression)}, execCtx_);
     } catch (const VeloxException& err) {
       std::cout << "Validation failed for expression in ProjectRel due to:"
                 << err.message() << std::endl;
@@ -228,6 +268,7 @@ bool SubstraitToVeloxPlanValidator::validate(
         auto typeCase = arg.rex_type_case();
         switch (typeCase) {
           case ::substrait::Expression::RexTypeCase::kSelection:
+          case ::substrait::Expression::RexTypeCase::kLiteral:
             break;
           default:
             std::cout << "Only field is supported in aggregate functions."
@@ -242,7 +283,8 @@ bool SubstraitToVeloxPlanValidator::validate(
     }
   }
 
-  std::unordered_set<std::string> supportedFuncs = {"sum", "count", "avg"};
+  std::unordered_set<std::string> supportedFuncs = {
+      "sum", "count", "avg", "min", "max"};
   for (const auto& funcSpec : funcSpecs) {
     auto funcName = subParser_->getSubFunctionName(funcSpec);
     if (supportedFuncs.find(funcName) == supportedFuncs.end()) {
@@ -256,110 +298,18 @@ bool SubstraitToVeloxPlanValidator::validate(
 
 bool SubstraitToVeloxPlanValidator::validate(
     const ::substrait::ReadRel& sRead) {
-  if (!sRead.has_base_schema()) {
-    std::cout << "Validation failed due to schema was not found in ReadRel."
+  try {
+    u_int32_t index;
+    std::vector<std::string> paths;
+    std::vector<u_int64_t> starts;
+    std::vector<u_int64_t> lengths;
+
+    planConverter_->toVeloxPlan(sRead, index, paths, starts, lengths);
+  } catch (const VeloxException& err) {
+    std::cout << "ReadRel validation failed due to:" << err.message()
               << std::endl;
     return false;
   }
-  const auto& sTypes = sRead.base_schema().struct_().types();
-  for (const auto& sType : sTypes) {
-    if (!validate(sType)) {
-      std::cout << "Validation failed due to type was not supported in ReadRel."
-                << std::endl;
-      return false;
-    }
-  }
-  std::vector<::substrait::Expression_ScalarFunction> scalarFunctions;
-  if (sRead.has_filter()) {
-    try {
-      planConverter_->flattenConditions(sRead.filter(), scalarFunctions);
-    } catch (const VeloxException& err) {
-      std::cout
-          << "Validation failed due to flattening conditions failed in ReadRel due to:"
-          << err.message() << std::endl;
-      return false;
-    }
-  }
-  // Get and validate the filter functions.
-  std::vector<std::string> funcSpecs;
-  funcSpecs.reserve(scalarFunctions.size());
-  for (const auto& scalarFunction : scalarFunctions) {
-    try {
-      funcSpecs.emplace_back(
-          planConverter_->findFuncSpec(scalarFunction.function_reference()));
-    } catch (const VeloxException& err) {
-      std::cout << "Validation failed in ReadRel due to:" << err.message()
-                << std::endl;
-      return false;
-    }
-
-    if (scalarFunction.args().size() == 1) {
-      // Field is expected.
-      for (const auto& param : scalarFunction.args()) {
-        auto typeCase = param.rex_type_case();
-        switch (typeCase) {
-          case ::substrait::Expression::RexTypeCase::kSelection:
-            break;
-          default:
-            std::cout << "Field is Expected." << std::endl;
-            return false;
-        }
-      }
-    } else if (scalarFunction.args().size() == 2) {
-      // Expect there being two args. One is field and the other is literal.
-      bool fieldExists = false;
-      bool litExists = false;
-      for (const auto& param : scalarFunction.args()) {
-        auto typeCase = param.rex_type_case();
-        switch (typeCase) {
-          case ::substrait::Expression::RexTypeCase::kSelection: {
-            fieldExists = true;
-            break;
-          }
-          case ::substrait::Expression::RexTypeCase::kLiteral: {
-            litExists = true;
-            break;
-          }
-          default:
-            std::cout << "Type case: " << typeCase
-                      << " is not supported in ReadRel." << std::endl;
-            return false;
-        }
-      }
-      if (!fieldExists || !litExists) {
-        std::cout << "Only the case of Field and Literal is supported."
-                  << std::endl;
-        return false;
-      }
-    } else {
-      std::cout << "More than two args is not supported in ReadRel."
-                << std::endl;
-      return false;
-    }
-  }
-  std::unordered_set<std::string> supportedFilters = {
-      "is_not_null", "gte", "gt", "lte", "lt"};
-  std::unordered_set<std::string> supportedTypes = {"opt", "req", "fp64"};
-  for (const auto& funcSpec : funcSpecs) {
-    // Validate the functions.
-    auto funcName = subParser_->getSubFunctionName(funcSpec);
-    if (supportedFilters.find(funcName) == supportedFilters.end()) {
-      std::cout << "Validation failed due to " << funcName
-                << " was not supported in ReadRel." << std::endl;
-      return false;
-    }
-
-    // Validate the types.
-    std::vector<std::string> funcTypes;
-    subParser_->getSubFunctionTypes(funcSpec, funcTypes);
-    for (const auto& funcType : funcTypes) {
-      if (supportedTypes.find(funcType) == supportedTypes.end()) {
-        std::cout << "Validation failed due to " << funcType
-                  << " was not supported in ReadRel." << std::endl;
-        return false;
-      }
-    }
-  }
   return true;
 }
 
@@ -393,10 +343,11 @@ bool SubstraitToVeloxPlanValidator::validate(
 
 bool SubstraitToVeloxPlanValidator::validate(const ::substrait::Plan& sPlan) {
   functions::prestosql::registerAllScalarFunctions();
+  functions::sparksql::registerFunctions("");
   // Create plan converter and expression converter to help the validation.
   planConverter_->constructFuncMap(sPlan);
   exprConverter_ = std::make_shared<SubstraitVeloxExprConverter>(
-      planConverter_->getFunctionMap());
+      pool_, planConverter_->getFunctionMap());
 
   for (const auto& sRel : sPlan.relations()) {
     if (sRel.has_root()) {
diff --git a/velox/substrait/SubstraitToVeloxPlanValidator.h b/velox/substrait/SubstraitToVeloxPlanValidator.h
index 52164696685b..dd6b8bca97ca 100644
--- a/velox/substrait/SubstraitToVeloxPlanValidator.h
+++ b/velox/substrait/SubstraitToVeloxPlanValidator.h
@@ -24,6 +24,11 @@ namespace facebook::velox::substrait {
 /// a Substrait plan is supported in Velox.
 class SubstraitToVeloxPlanValidator {
  public:
+  SubstraitToVeloxPlanValidator(
+      memory::MemoryPool* pool,
+      core::ExecCtx* execCtx)
+      : pool_(pool), execCtx_(execCtx) {}
+
   /// Used to validate whether the computing of this type is supported.
   bool validate(const ::substrait::Type& sType);
 
@@ -52,19 +57,15 @@ class SubstraitToVeloxPlanValidator {
   bool validate(const ::substrait::Plan& sPlan);
 
  private:
-  /// A query context used for function validation.
-  std::shared_ptr<core::QueryCtx> queryCtx_{core::QueryCtx::createForTest()};
-
   /// A memory pool used for function validation.
-  std::unique_ptr<memory::MemoryPool> pool_{
-      memory::getDefaultScopedMemoryPool()};
+  memory::MemoryPool* pool_;
 
   /// An execution context used for function validation.
-  core::ExecCtx execCtx_{pool_.get(), queryCtx_.get()};
+  core::ExecCtx* execCtx_;
 
   /// A converter used to convert Substrait plan into Velox's plan node.
   std::shared_ptr<SubstraitVeloxPlanConverter> planConverter_ =
-      std::make_shared<SubstraitVeloxPlanConverter>();
+      std::make_shared<SubstraitVeloxPlanConverter>(pool_, true);
 
   /// A parser used to convert Substrait plan into recognizable representations.
   std::shared_ptr<SubstraitParser> subParser_ =
diff --git a/velox/substrait/TypeUtils.h b/velox/substrait/TypeUtils.h
index 3a649eef674c..ab66848ba092 100644
--- a/velox/substrait/TypeUtils.h
+++ b/velox/substrait/TypeUtils.h
@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+#include "velox/type/Filter.h"
 #include "velox/type/Type.h"
 
 namespace facebook::velox::substrait {
@@ -24,4 +25,41 @@ TypePtr toVeloxType(const std::string& typeName);
 std::string_view getNameBeforeDelimiter(
     const std::string& compoundName,
     const std::string& delimiter);
+#ifndef RANGETRAITS_H
+#define RANGETRAITS_H
+
+// Traits used to map type kind to the range used in Filter.
+template <TypeKind KIND>
+struct RangeTraits {};
+
+template <>
+struct RangeTraits<TypeKind::INTEGER> {
+  using RangeType = common::BigintRange;
+  using MultiRangeType = common::BigintMultiRange;
+  using NativeType = int64_t;
+};
+
+template <>
+struct RangeTraits<TypeKind::BIGINT> {
+  using RangeType = common::BigintRange;
+  using MultiRangeType = common::BigintMultiRange;
+  using NativeType = int64_t;
+};
+
+template <>
+struct RangeTraits<TypeKind::DOUBLE> {
+  using RangeType = common::DoubleRange;
+  using MultiRangeType = common::MultiRange;
+  using NativeType = double;
+};
+
+template <>
+struct RangeTraits<TypeKind::VARCHAR> {
+  using RangeType = common::BytesRange;
+  using MultiRangeType = common::MultiRange;
+  using NativeType = std::string;
+};
+
+#endif /* RANGETRAITS_H */
+
 } // namespace facebook::velox::substrait
diff --git a/velox/substrait/VectorCreater.cpp b/velox/substrait/VectorCreater.cpp
new file mode 100644
index 000000000000..8ecd69e6f86b
--- /dev/null
+++ b/velox/substrait/VectorCreater.cpp
@@ -0,0 +1,94 @@
+/*
+ * 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/substrait/VectorCreater.h"
+#include "velox/vector/FlatVector.h"
+
+namespace facebook::velox::substrait {
+
+template <TypeKind KIND>
+VectorPtr setVectorFromVariantsByKind(
+    const std::vector<velox::variant>& value,
+    memory::MemoryPool* pool) {
+  using T = typename TypeTraits<KIND>::NativeType;
+
+  auto flatVector = std::dynamic_pointer_cast<FlatVector<T>>(
+      BaseVector::create(CppToType<T>::create(), value.size(), pool));
+
+  for (vector_size_t i = 0; i < value.size(); i++) {
+    if (value[i].isNull()) {
+      flatVector->setNull(i, true);
+    } else {
+      flatVector->set(i, value[i].value<T>());
+    }
+  }
+  return flatVector;
+}
+
+template <>
+VectorPtr setVectorFromVariantsByKind<TypeKind::VARBINARY>(
+    const std::vector<velox::variant>& value,
+    memory::MemoryPool* pool) {
+  throw std::invalid_argument("Return of VARBINARY data is not supported");
+}
+
+template <>
+VectorPtr setVectorFromVariantsByKind<TypeKind::VARCHAR>(
+    const std::vector<velox::variant>& value,
+    memory::MemoryPool* pool) {
+  auto flatVector = std::dynamic_pointer_cast<FlatVector<StringView>>(
+      BaseVector::create(VARCHAR(), value.size(), pool));
+
+  for (vector_size_t i = 0; i < value.size(); i++) {
+    if (value[i].isNull()) {
+      flatVector->setNull(i, true);
+    } else {
+      flatVector->set(i, StringView(value[i].value<Varchar>()));
+    }
+  }
+  return flatVector;
+}
+
+VectorPtr setVectorFromVariants(
+    const TypePtr& type,
+    const std::vector<velox::variant>& value,
+    memory::MemoryPool* pool) {
+  return VELOX_DYNAMIC_SCALAR_TYPE_DISPATCH(
+      setVectorFromVariantsByKind, type->kind(), value, pool);
+}
+
+ArrayVectorPtr
+toArrayVector(TypePtr type, VectorPtr vector, memory::MemoryPool* pool) {
+  vector_size_t size = 1;
+  BufferPtr offsets = AlignedBuffer::allocate<vector_size_t>(size, pool);
+  BufferPtr sizes = AlignedBuffer::allocate<vector_size_t>(size, pool);
+  BufferPtr nulls = AlignedBuffer::allocate<uint64_t>(size, pool);
+
+  auto rawOffsets = offsets->asMutable<vector_size_t>();
+  auto rawSizes = sizes->asMutable<vector_size_t>();
+  auto rawNulls = nulls->asMutable<uint64_t>();
+
+  bits::fillBits(rawNulls, 0, size, pool);
+  vector_size_t nullCount = 0;
+
+  *rawSizes++ = vector->size();
+  *rawOffsets++ = 0;
+
+  return std::make_shared<ArrayVector>(
+      pool, ARRAY(type), nulls, size, offsets, sizes, vector, nullCount);
+}
+
+} // namespace facebook::velox::substrait
diff --git a/velox/substrait/VectorCreater.h b/velox/substrait/VectorCreater.h
new file mode 100644
index 000000000000..36fd81212773
--- /dev/null
+++ b/velox/substrait/VectorCreater.h
@@ -0,0 +1,37 @@
+/*
+ * 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/vector/ComplexVector.h"
+
+namespace facebook::velox::substrait {
+
+/// Create different types of vector from velox variants.
+template <TypeKind KIND>
+VectorPtr setVectorFromVariantsByKind(
+    const std::vector<velox::variant>& value,
+    memory::MemoryPool* pool);
+
+/// Create Vector from velox variants.
+VectorPtr setVectorFromVariants(
+    const TypePtr& type,
+    const std::vector<velox::variant>& value,
+    velox::memory::MemoryPool* pool);
+
+/// Create Array Vector from Vector.
+ArrayVectorPtr
+toArrayVector(TypePtr type, VectorPtr vector, memory::MemoryPool* pool);
+
+} // namespace facebook::velox::substrait
diff --git a/velox/substrait/tests/PlanConversionTest.cpp b/velox/substrait/tests/PlanConversionTest.cpp
index 7f7d0fe56d45..74bd99b07c9f 100644
--- a/velox/substrait/tests/PlanConversionTest.cpp
+++ b/velox/substrait/tests/PlanConversionTest.cpp
@@ -43,22 +43,6 @@ using namespace facebook::velox::exec::test;
 class PlanConversionTest : public virtual HiveConnectorTestBase,
                            public testing::WithParamInterface<bool> {
  protected:
-  void SetUp() override {
-    useAsyncCache_ = GetParam();
-    HiveConnectorTestBase::SetUp();
-  }
-
-  static void SetUpTestCase() {
-    HiveConnectorTestBase::SetUpTestCase();
-  }
-
-  std::vector<RowVectorPtr> makeVectors(
-      int32_t count,
-      int32_t rowsPerVector,
-      const std::shared_ptr<const RowType>& rowType) {
-    return HiveConnectorTestBase::makeVectors(rowType, count, rowsPerVector);
-  }
-
   class VeloxConverter {
    public:
     // This class is an iterator for Velox computing.
@@ -165,7 +149,8 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
       google::protobuf::util::JsonStringToMessage(subData, &subPlan);
 
       auto planConverter = std::make_shared<
-          facebook::velox::substrait::SubstraitVeloxPlanConverter>();
+          facebook::velox::substrait::SubstraitVeloxPlanConverter>(
+          memoryPool_.get());
       // Convert to Velox PlanNode.
       auto planNode = planConverter->toVeloxPlan(subPlan);
 
@@ -196,23 +181,46 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
 
     std::shared_ptr<exec::test::TempDirectoryPath> tmpDir_{
         exec::test::TempDirectoryPath::create()};
+
+   private:
+    std::unique_ptr<memory::MemoryPool> memoryPool_{
+        memory::getDefaultScopedMemoryPool()};
   };
 
+  void SetUp() override {
+    useAsyncCache_ = GetParam();
+    HiveConnectorTestBase::SetUp();
+  }
+
+  static void SetUpTestCase() {
+    HiveConnectorTestBase::SetUpTestCase();
+  }
+
+  std::vector<RowVectorPtr> makeVectors(
+      int32_t count,
+      int32_t rowsPerVector,
+      const std::shared_ptr<const RowType>& rowType) {
+    return HiveConnectorTestBase::makeVectors(rowType, count, rowsPerVector);
+  }
+
+  std::unique_ptr<memory::MemoryPool> pool_{
+      facebook::velox::memory::getDefaultScopedMemoryPool()};
+
   // This method can be used to create a Fixed-width type of Vector without Null
   // values.
   template <typename T>
   VectorPtr createSpecificScalar(
       size_t size,
       std::vector<T> vals,
-      facebook::velox::memory::MemoryPool& pool) {
-    facebook::velox::BufferPtr values = AlignedBuffer::allocate<T>(size, &pool);
+      memory::MemoryPool* pool) {
+    facebook::velox::BufferPtr values = AlignedBuffer::allocate<T>(size, pool);
     auto valuesPtr = values->asMutableRange<T>();
     facebook::velox::BufferPtr nulls = nullptr;
     for (size_t i = 0; i < size; ++i) {
       valuesPtr[i] = vals[i];
     }
     return std::make_shared<facebook::velox::FlatVector<T>>(
-        &pool, nulls, size, values, std::vector<BufferPtr>{});
+        pool, nulls, size, values, std::vector<BufferPtr>{});
   }
 
   // This method can be used to create a String type of Vector without Null
@@ -220,8 +228,8 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
   VectorPtr createSpecificStringVector(
       size_t size,
       std::vector<std::string> vals,
-      facebook::velox::memory::MemoryPool& pool) {
-    auto vector = BaseVector::create(VARCHAR(), size, &pool);
+      memory::MemoryPool* pool) {
+    auto vector = BaseVector::create(VARCHAR(), size, pool);
     auto flatVector = vector->asFlatVector<StringView>();
 
     size_t childSize = 0;
@@ -236,7 +244,7 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
     }
     vector->setNullCount(0);
 
-    BufferPtr buf = AlignedBuffer::allocate<char>(childSize, &pool);
+    BufferPtr buf = AlignedBuffer::allocate<char>(childSize, pool);
     char* bufPtr = buf->asMutable<char>();
     char* dest = bufPtr;
     for (size_t i = 0; i < size; ++i) {
@@ -291,8 +299,7 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
              VARCHAR(),
              VARCHAR(),
              VARCHAR()});
-    std::unique_ptr<facebook::velox::memory::MemoryPool> pool{
-        facebook::velox::memory::getDefaultScopedMemoryPool()};
+
     std::vector<VectorPtr> vectors;
     // TPC-H lineitem table has 16 columns.
     int colNum = 16;
@@ -309,7 +316,7 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
         6354246853,
         4141748419};
     vectors.emplace_back(
-        createSpecificScalar<int64_t>(10, lOrderkeyData, *pool));
+        createSpecificScalar<int64_t>(10, lOrderkeyData, pool_.get()));
     std::vector<int64_t> lPartkeyData = {
         263222018,
         255918298,
@@ -322,7 +329,7 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
         112999357,
         299103530};
     vectors.emplace_back(
-        createSpecificScalar<int64_t>(10, lPartkeyData, *pool));
+        createSpecificScalar<int64_t>(10, lPartkeyData, pool_.get()));
     std::vector<int64_t> lSuppkeyData = {
         2102019,
         13998315,
@@ -335,14 +342,14 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
         1639379,
         3423588};
     vectors.emplace_back(
-        createSpecificScalar<int64_t>(10, lSuppkeyData, *pool));
+        createSpecificScalar<int64_t>(10, lSuppkeyData, pool_.get()));
     std::vector<int32_t> lLinenumberData = {4, 6, 1, 5, 1, 2, 1, 5, 2, 6};
     vectors.emplace_back(
-        createSpecificScalar<int32_t>(10, lLinenumberData, *pool));
+        createSpecificScalar<int32_t>(10, lLinenumberData, pool_.get()));
     std::vector<double> lQuantityData = {
         6.0, 1.0, 19.0, 4.0, 6.0, 12.0, 23.0, 11.0, 16.0, 19.0};
     vectors.emplace_back(
-        createSpecificScalar<double>(10, lQuantityData, *pool));
+        createSpecificScalar<double>(10, lQuantityData, pool_.get()));
     std::vector<double> lExtendedpriceData = {
         30586.05,
         7821.0,
@@ -355,22 +362,23 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
         8704.26,
         63780.36};
     vectors.emplace_back(
-        createSpecificScalar<double>(10, lExtendedpriceData, *pool));
+        createSpecificScalar<double>(10, lExtendedpriceData, pool_.get()));
     std::vector<double> lDiscountData = {
         0.05, 0.06, 0.01, 0.07, 0.05, 0.06, 0.07, 0.05, 0.06, 0.07};
     vectors.emplace_back(
-        createSpecificScalar<double>(10, lDiscountData, *pool));
+        createSpecificScalar<double>(10, lDiscountData, pool_.get()));
     std::vector<double> lTaxData = {
         0.02, 0.03, 0.01, 0.0, 0.01, 0.01, 0.03, 0.07, 0.01, 0.04};
-    vectors.emplace_back(createSpecificScalar<double>(10, lTaxData, *pool));
+    vectors.emplace_back(
+        createSpecificScalar<double>(10, lTaxData, pool_.get()));
     std::vector<std::string> lReturnflagData = {
         "N", "A", "A", "R", "A", "N", "A", "A", "N", "R"};
     vectors.emplace_back(
-        createSpecificStringVector(10, lReturnflagData, *pool));
+        createSpecificStringVector(10, lReturnflagData, pool_.get()));
     std::vector<std::string> lLinestatusData = {
         "O", "F", "F", "F", "F", "O", "F", "F", "O", "F"};
     vectors.emplace_back(
-        createSpecificStringVector(10, lLinestatusData, *pool));
+        createSpecificStringVector(10, lLinestatusData, pool_.get()));
     std::vector<double> lShipdateNewData = {
         8953.666666666666,
         8773.666666666666,
@@ -383,7 +391,7 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
         9013.666666666666,
         8832.666666666666};
     vectors.emplace_back(
-        createSpecificScalar<double>(10, lShipdateNewData, *pool));
+        createSpecificScalar<double>(10, lShipdateNewData, pool_.get()));
     std::vector<double> lCommitdateNewData = {
         10447.666666666666,
         8953.666666666666,
@@ -396,7 +404,7 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
         9519.666666666666,
         9138.666666666666};
     vectors.emplace_back(
-        createSpecificScalar<double>(10, lCommitdateNewData, *pool));
+        createSpecificScalar<double>(10, lCommitdateNewData, pool_.get()));
     std::vector<double> lReceiptdateNewData = {
         10456.666666666666,
         8979.666666666666,
@@ -409,7 +417,7 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
         9593.666666666666,
         9178.666666666666};
     vectors.emplace_back(
-        createSpecificScalar<double>(10, lReceiptdateNewData, *pool));
+        createSpecificScalar<double>(10, lReceiptdateNewData, pool_.get()));
     std::vector<std::string> lShipinstructData = {
         "COLLECT COD",
         "NONE",
@@ -422,7 +430,7 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
         "TAKE BACK RETURN",
         "NONE"};
     vectors.emplace_back(
-        createSpecificStringVector(10, lShipinstructData, *pool));
+        createSpecificStringVector(10, lShipinstructData, pool_.get()));
     std::vector<std::string> lShipmodeData = {
         "FOB",
         "REG AIR",
@@ -434,7 +442,8 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
         "REG AIR",
         "TRUCK",
         "AIR"};
-    vectors.emplace_back(createSpecificStringVector(10, lShipmodeData, *pool));
+    vectors.emplace_back(
+        createSpecificStringVector(10, lShipmodeData, pool_.get()));
     std::vector<std::string> lCommentData = {
         " the furiously final foxes. quickly final p",
         "thely ironic",
@@ -446,12 +455,13 @@ class PlanConversionTest : public virtual HiveConnectorTestBase,
         "lyly regular excuses affi",
         "lly unusual theodolites grow slyly above",
         " the quickly ironic pains lose car"};
-    vectors.emplace_back(createSpecificStringVector(10, lCommentData, *pool));
+    vectors.emplace_back(
+        createSpecificStringVector(10, lCommentData, pool_.get()));
 
     // Batches has only one RowVector here.
     uint64_t nullCount = 0;
     std::vector<RowVectorPtr> batches{std::make_shared<RowVector>(
-        pool.get(), type, nullptr, 10, vectors, nullCount)};
+        pool_.get(), type, nullptr, 10, vectors, nullCount)};
 
     // Writes data into an ORC file.
     auto sink = std::make_unique<facebook::velox::dwio::common::FileSink>(
diff --git a/velox/type/Filter.cpp b/velox/type/Filter.cpp
index 0d516ebd118a..7c25ca4cea35 100644
--- a/velox/type/Filter.cpp
+++ b/velox/type/Filter.cpp
@@ -65,6 +65,9 @@ std::string Filter::toString() const {
     case FilterKind::kDoubleRange:
       strKind = "DoubleRange";
       break;
+    case FilterKind::kDoubleValues:
+      strKind = "DoubleValues";
+      break;
     case FilterKind::kFloatRange:
       strKind = "FloatRange";
       break;
@@ -208,6 +211,103 @@ bool BigintValuesUsingHashTable::testInt64(int64_t value) const {
   return false;
 }
 
+DoubleValues::DoubleValues(
+    double min,
+    double max,
+    const std::vector<double>& values,
+    bool nullAllowed)
+    : Filter(true, nullAllowed, FilterKind::kDoubleValues),
+      min_(min),
+      max_(max) {
+  VELOX_CHECK(min < max, "min must be less than max");
+  VELOX_CHECK(values.size() > 1, "values must contain at least 2 entries");
+
+  bitmask_.resize(toInt64(max) - toInt64(min) + 1);
+
+  for (double value : values) {
+    bitmask_[toInt64(value) - toInt64(min)] = true;
+  }
+}
+
+bool DoubleValues::testDouble(double value) const {
+  if (value < min_ || value > max_) {
+    return false;
+  }
+  return bitmask_[toInt64(value) - toInt64(min_)];
+}
+
+std::vector<double> DoubleValues::values() const {
+  std::vector<double> values;
+  for (int i = 0; i < bitmask_.size(); i++) {
+    if (bitmask_[i]) {
+      values.push_back(min_ + i);
+    }
+  }
+  return values;
+}
+
+bool DoubleValues::testDoubleRange(double min, double max, bool hasNull) const {
+  if (hasNull && nullAllowed_) {
+    return true;
+  }
+
+  if (toInt64(min) == toInt64(max)) {
+    return testDouble(min);
+  }
+
+  return !(min > max_ || max < min_);
+}
+
+std::unique_ptr<Filter> DoubleValues::mergeWith(const Filter* other) const {
+  switch (other->kind()) {
+    case FilterKind::kAlwaysTrue:
+    case FilterKind::kAlwaysFalse:
+    case FilterKind::kIsNull:
+      return other->mergeWith(this);
+    case FilterKind::kIsNotNull:
+      return std::make_unique<DoubleValues>(*this, false);
+    case FilterKind::kDoubleRange: {
+      auto otherRange = dynamic_cast<const DoubleRange*>(other);
+
+      auto min = std::max(min_, otherRange->lower());
+      auto max = std::min(max_, otherRange->upper());
+
+      return mergeWith(min, max, other);
+    }
+    case FilterKind::kFloatRange: {
+      auto otherRange = dynamic_cast<const FloatRange*>(other);
+
+      auto min = std::max(min_, otherRange->lower());
+      auto max = std::min(max_, otherRange->upper());
+
+      return mergeWith(min, max, other);
+    }
+    case FilterKind::kDoubleValues: {
+      auto otherValues = dynamic_cast<const DoubleValues*>(other);
+
+      auto min = std::max(min_, otherValues->min_);
+      auto max = std::min(max_, otherValues->max_);
+
+      return mergeWith(min, max, other);
+    }
+    default:
+      VELOX_UNREACHABLE();
+  }
+}
+
+std::unique_ptr<Filter>
+DoubleValues::mergeWith(double min, double max, const Filter* other) const {
+  bool bothNullAllowed = nullAllowed_ && other->testNull();
+
+  std::vector<double> valuesToKeep;
+  for (auto i = min; i <= max; ++i) {
+    if (bitmask_[toInt64(i) - toInt64(min_)] && other->testDouble(i)) {
+      valuesToKeep.push_back(i);
+    }
+  }
+  return createDoubleValues(valuesToKeep, bothNullAllowed);
+}
+
 xsimd::batch_bool<int64_t> BigintValuesUsingHashTable::testValues(
     xsimd::batch<int64_t> x) const {
   auto outOfRange = (x < xsimd::broadcast<int64_t>(min_)) |
@@ -396,6 +496,31 @@ std::unique_ptr<Filter> notNullOrTrue(bool nullAllowed) {
   return std::make_unique<IsNotNull>();
 }
 
+std::unique_ptr<Filter> createDoubleValues(
+    const std::vector<double>& values,
+    bool nullAllowed) {
+  if (values.empty()) {
+    return nullOrFalse(nullAllowed);
+  }
+  if (values.size() == 1) {
+    return std::make_unique<DoubleRange>(
+        values.front(),
+        false,
+        false,
+        values.front(),
+        false,
+        false,
+        nullAllowed);
+  }
+  double min = values.front();
+  double max = values.front();
+  for (const auto& value : values) {
+    min = (value < min) ? value : min;
+    max = (value > max) ? value : max;
+  }
+  return std::make_unique<DoubleValues>(min, max, values, nullAllowed);
+}
+
 std::unique_ptr<Filter> createBigintValuesFilter(
     const std::vector<int64_t>& values,
     bool nullAllowed,
diff --git a/velox/type/Filter.h b/velox/type/Filter.h
index 31f2986df865..4b7adf3f5bea 100644
--- a/velox/type/Filter.h
+++ b/velox/type/Filter.h
@@ -45,6 +45,7 @@ enum class FilterKind {
   kNegatedBigintValuesUsingHashTable,
   kNegatedBigintValuesUsingBitmask,
   kDoubleRange,
+  kDoubleValues,
   kFloatRange,
   kBytesRange,
   kNegatedBytesRange,
@@ -554,6 +555,23 @@ class BigintRange final : public Filter {
         upper16_(std::min<int64_t>(upper, std::numeric_limits<int16_t>::max())),
         isSingleValue_(upper_ == lower_) {}
 
+  BigintRange(
+      int64_t lower,
+      bool lowerUnbounded,
+      bool lowerExclusive,
+      int64_t upper,
+      bool upperUnbounded,
+      bool upperExclusive,
+      bool nullAllowed)
+      : Filter(true, nullAllowed, FilterKind::kBigintRange),
+        lower_(lowerExclusive ? lower - 1 : lower),
+        upper_(upperExclusive ? upper + 1 : upper),
+        lower32_(std::max<int64_t>(lower, std::numeric_limits<int32_t>::min())),
+        upper32_(std::min<int64_t>(upper, std::numeric_limits<int32_t>::max())),
+        lower16_(std::max<int64_t>(lower, std::numeric_limits<int16_t>::min())),
+        upper16_(std::min<int64_t>(upper, std::numeric_limits<int16_t>::max())),
+        isSingleValue_(upper_ == lower_) {}
+
   std::unique_ptr<Filter> clone(
       std::optional<bool> nullAllowed = std::nullopt) const final {
     if (nullAllowed) {
@@ -911,8 +929,6 @@ class NegatedBigintValuesUsingHashTable final : public Filter {
   std::unique_ptr<BigintValuesUsingHashTable> nonNegated_;
 };
 
-/// NOT IN-list filter for integral data types. Implemented as a bitmask. Offers
-/// better performance than the hash table when the range of values is small.
 class NegatedBigintValuesUsingBitmask final : public Filter {
  public:
   /// @param min Minimum REJECTED value.
@@ -962,6 +978,56 @@ class NegatedBigintValuesUsingBitmask final : public Filter {
   std::unique_ptr<BigintValuesUsingBitmask> nonNegated_;
 };
 
+class DoubleValues final : public Filter {
+ public:
+  /// @param min Minimum value.
+  /// @param max Maximum value.
+  /// @param values A list of unique values that pass the filter. Must contain
+  /// at least two entries.
+  /// @param nullAllowed Null values are passing the filter if true.
+  DoubleValues(
+      double min,
+      double max,
+      const std::vector<double>& values,
+      bool nullAllowed);
+
+  DoubleValues(const DoubleValues& other, bool nullAllowed)
+      : Filter(true, nullAllowed, FilterKind::kDoubleValues),
+        bitmask_(other.bitmask_),
+        min_(other.min_),
+        max_(other.max_) {}
+
+  std::unique_ptr<Filter> clone(
+      std::optional<bool> nullAllowed = std::nullopt) const final {
+    if (nullAllowed) {
+      return std::make_unique<DoubleValues>(*this, nullAllowed.value());
+    } else {
+      return std::make_unique<DoubleValues>(*this);
+    }
+  }
+
+  std::vector<double> values() const;
+
+  bool testDouble(double value) const final;
+
+  bool testDoubleRange(double min, double max, bool hasNull) const final;
+
+  std::unique_ptr<Filter> mergeWith(const Filter* other) const final;
+
+ private:
+  std::unique_ptr<Filter> mergeWith(double min, double max, const Filter* other)
+      const;
+
+  int64_t toInt64(double value) const {
+    int64_t converted = (int64_t)(value + 0.5);
+    return converted;
+  }
+
+  std::vector<bool> bitmask_;
+  const double min_;
+  const double max_;
+};
+
 /// Base class for range filters on floating point and string data types.
 class AbstractRange : public Filter {
  public:
@@ -1639,6 +1705,11 @@ class MultiRange final : public Filter {
         filters_(std::move(filters)),
         nanAllowed_(nanAllowed) {}
 
+  MultiRange(std::vector<std::unique_ptr<Filter>> filters, bool nullAllowed)
+      : Filter(true, nullAllowed, FilterKind::kMultiRange),
+        filters_(std::move(filters)),
+        nanAllowed_(true) {}
+
   std::unique_ptr<Filter> clone(
       std::optional<bool> nullAllowed = std::nullopt) const final;
 
@@ -1714,4 +1785,8 @@ std::unique_ptr<Filter> createNegatedBigintValues(
     const std::vector<int64_t>& values,
     bool nullAllowed);
 
+std::unique_ptr<Filter> createDoubleValues(
+    const std::vector<double>& values,
+    bool nullAllowed);
+
 } // namespace facebook::velox::common