align validations

Author: rui-mo

velox/substrait/CMakeLists.txt

@@ -47,7 +47,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)

velox/substrait/SubstraitToVeloxPlan.cpp

@@ -549,12 +549,10 @@ std::shared_ptr<const core::PlanNode> SubstraitVeloxPlanConverter::toVeloxPlan(
 
 std::shared_ptr<const core::PlanNode> SubstraitVeloxPlanConverter::toVeloxPlan(
     const ::substrait::Plan& sPlan) {
-  // Construct the function map based on the Substrait representation.
+  // Construct the function map based on the Substrait representation,
+  // and initialize the expression converter with it.
   constructFuncMap(sPlan);
 
-  // Create the expression converter.
-  exprConverter_ = std::make_shared<SubstraitVeloxExprConverter>(functionMap_);
-
   // In fact, only one RelRoot or Rel is expected here.
   for (const auto& sRel : sPlan.relations()) {
     if (sRel.has_root()) {
@@ -579,6 +577,7 @@ void SubstraitVeloxPlanConverter::constructFuncMap(
     auto name = sFmap.name();
     functionMap_[id] = name;
   }
+  exprConverter_ = std::make_shared<SubstraitVeloxExprConverter>(functionMap_);
 }
 
 std::string SubstraitVeloxPlanConverter::nextPlanNodeId() {
@@ -674,6 +673,9 @@ int32_t SubstraitVeloxPlanConverter::streamIsInput(
       VELOX_FAIL(err.what());
     }
   }
+  if (validationMode_) {
+    return -1;
+  }
   VELOX_FAIL("Local file is expected.");
 }
 
@@ -1316,7 +1318,7 @@ SubstraitVeloxPlanConverter::connectWithAnd(
   while (idx < remainingFunctions.size()) {
     std::vector<std::shared_ptr<const core::ITypedExpr>> params;
     params.reserve(2);
-    params.emplace_back(std::move(remainingFilter));
+    params.emplace_back(remainingFilter);
     params.emplace_back(
         exprConverter_->toVeloxExpr(remainingFunctions[idx], inputType));
     remainingFilter = std::make_shared<const core::CallTypedExpr>(

velox/substrait/SubstraitToVeloxPlan.h

@@ -25,6 +25,9 @@ namespace facebook::velox::substrait {
 /// This class is used to convert the Substrait plan into Velox plan.
 class SubstraitVeloxPlanConverter {
  public:
+  SubstraitVeloxPlanConverter(bool validationMode = false)
+      : validationMode_(validationMode) {}
+
   /// Used to convert Substrait JoinRel into Velox PlanNode.
   std::shared_ptr<const core::PlanNode> toVeloxPlan(
       const ::substrait::JoinRel& sJoin);
@@ -65,7 +68,8 @@ class SubstraitVeloxPlanConverter {
       const ::substrait::Plan& sPlan);
 
   /// Used to construct the function map between the index
-  /// and the Substrait function name.
+  /// and the Substrait function name. Initialize the expression
+  /// converter based on the constructed function map.
   void constructFuncMap(const ::substrait::Plan& sPlan);
 
   /// Will return the function map used by this plan converter.
@@ -371,6 +375,9 @@ class SubstraitVeloxPlanConverter {
   /// The Expression converter used to convert Substrait representations into
   /// Velox expressions.
   std::shared_ptr<SubstraitVeloxExprConverter> exprConverter_;
+
+  /// A flag used to specify validation.
+  bool validationMode_ = false;
 };
 
 } // namespace facebook::velox::substrait

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;
   }
 
@@ -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(
@@ -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."
@@ -256,110 +297,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,6 +342,7 @@ 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>(

velox/substrait/SubstraitToVeloxPlanValidator.h

@@ -64,7 +64,7 @@ class SubstraitToVeloxPlanValidator {
 
   /// 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>(true);
 
   /// A parser used to convert Substrait plan into recognizable representations.
   std::shared_ptr<SubstraitParser> subParser_ =