Add more validations for functions (#198)

oap-project · Apr 21, 2023 · 10addd0 · 10addd0
1 parent f265f03
commit 10addd0
Show file tree

Hide file tree

Showing 5 changed files with 61 additions and 16 deletions.
diff --git a/velox/common/file/FileSystems.h b/velox/common/file/FileSystems.h
@@ -89,8 +89,9 @@ std::shared_ptr<FileSystem> getFileSystem(
 // and a lambda that generates the actual file system.
 void registerFileSystem(
     std::function<bool(std::string_view)> schemeMatcher,
-    std::function<std::shared_ptr<FileSystem>(std::shared_ptr<const Config>, std::string_view)>
-        fileSystemGenerator);
+    std::function<std::shared_ptr<FileSystem>(
+        std::shared_ptr<const Config>,
+        std::string_view)> fileSystemGenerator);
 
 // Register the local filesystem.
 void registerLocalFileSystem();

diff --git a/velox/functions/sparksql/aggregates/DecimalAvgAggregate.h b/velox/functions/sparksql/aggregates/DecimalAvgAggregate.h
@@ -297,12 +297,12 @@ class DecimalAverageAggregate : public exec::Aggregate {
     if ((accumulator->overflow == 1 && accumulator->sum < 0) ||
         (accumulator->overflow == -1 && accumulator->sum > 0)) {
       sum = static_cast<int128_t>(
-              DecimalUtil::kOverflowMultiplier * accumulator->overflow +
-                      accumulator->sum);
+          DecimalUtil::kOverflowMultiplier * accumulator->overflow +
+          accumulator->sum);
     } else {
       VELOX_CHECK(
-              accumulator->overflow == 0,
-              "overflow: decimal avg struct overflow not eq 0");
+          accumulator->overflow == 0,
+          "overflow: decimal avg struct overflow not eq 0");
     }
 
     auto [resultPrecision, resultScale] =
@@ -336,12 +336,7 @@ class DecimalAverageAggregate : public exec::Aggregate {
           UnscaledLongDecimal,
           UnscaledLongDecimal,
           UnscaledLongDecimal>(
-          avg,
-          UnscaledLongDecimal(sum),
-          countDecimal,
-          false,
-          sumRescale,
-          0);
+          avg, UnscaledLongDecimal(sum), countDecimal, false, sumRescale, 0);
     }
     auto castedAvg =
         DecimalUtil::rescaleWithRoundUp<UnscaledLongDecimal, TResultType>(

diff --git a/velox/row/UnsafeRowDeserializer.h b/velox/row/UnsafeRowDeserializer.h
@@ -1037,7 +1037,8 @@ struct UnsafeRowDynamicVectorDeserializer {
   static VectorPtr convertPrimitiveIteratorsToVectors(
       std::vector<DataIteratorPtr>::iterator dataIterators,
       memory::MemoryPool* pool,
-      size_t numIteratorsToProcess,int32_t numFields = 1,
+      size_t numIteratorsToProcess,
+      int32_t numFields = 1,
       int fieldsIdx = 0) {
     TypePtr type = (*dataIterators)->type();
     assert(type->isPrimitiveType());

diff --git a/velox/row/tests/UnsafeRowBatchDeserializerTest.cpp b/velox/row/tests/UnsafeRowBatchDeserializerTest.cpp
@@ -665,7 +665,11 @@ class UnsafeRowComplexBatchDeserializerTests
           return StringView::makeInline("str" + std::to_string(row + index));
         });
     return makeRowVector(
-        {intVector, stringVector, decimalVector, intArrayVector, stringArrayVector});
+        {intVector,
+         stringVector,
+         decimalVector,
+         intArrayVector,
+         stringArrayVector});
   }
 
   std::shared_ptr<memory::MemoryPool> pool_ = memory::getDefaultMemoryPool();

diff --git a/velox/substrait/SubstraitToVeloxPlanValidator.cpp b/velox/substrait/SubstraitToVeloxPlanValidator.cpp
@@ -95,11 +95,44 @@ bool SubstraitToVeloxPlanValidator::validateRound(
 bool SubstraitToVeloxPlanValidator::validateScalarFunction(
     const ::substrait::Expression::ScalarFunction& scalarFunction,
     const RowTypePtr& inputType) {
-  const auto& veloxFunction = subParser_->findVeloxFunction(
+  const auto& function = subParser_->findSubstraitFuncSpec(
       planConverter_->getFunctionMap(), scalarFunction.function_reference());
-  if (veloxFunction == "round") {
+  const auto& name = subParser_->getSubFunctionName(function);
+  std::vector<std::string> types;
+  subParser_->getSubFunctionTypes(function, types);
+  if (name == "round") {
     return validateRound(scalarFunction, inputType);
   }
+  if (name == "char_length") {
+    VELOX_CHECK(types.size() == 1);
+    if (types[0] == "vbin") {
+      VLOG(1) << "Binary type is not supported in " << name << ".";
+      return false;
+    }
+  }
+  std::unordered_set<std::string> functions = {
+      "regexp_replace",
+      "split",
+      "split_part",
+      "factorial",
+      "concat_ws",
+      "rand",
+      "json_array_length",
+      "from_unixtime",
+      "to_unix_timestamp",
+      "unix_timestamp",
+      "repeat",
+      "translate",
+      "add_months",
+      "date_format",
+      "trunc",
+      "sequence",
+      "posexplode"};
+  if (functions.find(name) != functions.end()) {
+    VLOG(1) << "Function is not supported: " << name << ".";
+    return false;
+  }
+
   return true;
 }
 
@@ -330,6 +363,17 @@ bool SubstraitToVeloxPlanValidator::validate(
     }
   }
 
+  // Validate supported aggregate functions.
+  std::unordered_set<std::string> unsupportedFuncs = {"collect_list"};
+  for (const auto& funcSpec : funcSpecs) {
+    auto funcName = subParser_->getSubFunctionName(funcSpec);
+    if (unsupportedFuncs.find(funcName) != unsupportedFuncs.end()) {
+      std::cout << "Validation failed due to " << funcName
+                << " was not supported in WindowRel." << std::endl;
+      return false;
+    }
+  }
+
   // Validate groupby expression
   const auto& groupByExprs = windowRel.partition_expressions();
   std::vector<std::shared_ptr<const core::ITypedExpr>> expressions;