Skip to content
Open
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,8 @@ import org.apache.gluten.substrait.plan.PlanNode
import org.apache.gluten.validate.NativePlanValidationInfo
import org.apache.gluten.vectorized.NativePlanEvaluator

import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.expressions.{Attribute, EmptyRow, Expression, Round}
import org.apache.spark.sql.catalyst.plans.physical.Partitioning
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.types._
Expand All @@ -38,12 +39,32 @@ import io.substrait.proto.SimpleExtensionDeclaration
import scala.collection.JavaConverters._
import scala.collection.mutable.ArrayBuffer

class VeloxValidatorApi extends ValidatorApi {
class VeloxValidatorApi extends ValidatorApi with Logging {
import VeloxValidatorApi._

/** For velox backend, key validation is on native side. */
override def doExprValidate(substraitExprName: String, expr: Expression): Boolean =
true
/**
* Enforce decimal ROUND scale compatibility on the JVM independently of optional native plan
* validation.
*/
override def doExprValidate(substraitExprName: String, expr: Expression): Boolean = expr match {
case round: Round if round.child.dataType.isInstanceOf[DecimalType] =>

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This validation might be duplicate, as we've added a native validation in the "SubstraitToVeloxPlanValidator.cc" (which is a preferred way with a unified validate API).

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@rui-mo, the bounds overlap, but these checks are not equivalent under the supported validation settings. spark.gluten.sql.enable.native.validation=false skips the JNI/Substrait plan validator. The Spark-expression eligibility hook still runs while constructing the candidate Substrait plan:

  • Mandatory expression eligibility:
    private def getAndCheckSubstraitName(
    expr: Expression,
    expressionsMap: Map[Class[_], String]): String = {
    TestStats.addExpressionClassName(expr.getClass.getName)
    // Check whether Gluten supports this expression
    expressionsMap
    .get(expr.getClass)
    .filter(BackendsApiManager.getValidatorApiInstance.doExprValidate(_, expr))
    .getOrElse {
    throw new GlutenNotSupportException(
    s"Not supported to map spark function name" +
    s" to substrait function name: $expr, class name: ${expr.getClass.getSimpleName}.")
    }
    }
  • Optional native plan validation:
    protected def doNativeValidation(context: SubstraitContext, node: RelNode): ValidationResult = {
    if (node != null && enableNativeValidation) {
    val planNode = PlanBuilder.makePlan(context, Lists.newArrayList(node))
    BackendsApiManager.getValidatorApiInstance
    .doNativeValidateWithFailureReason(planNode)
    } else {
    ValidationResult.succeeded
    }
    }

Without the JVM eligibility guard, that configuration could offload an unsupported scale and fail in native execution instead of retaining Spark. For example, Spark 4.1.1 successfully evaluates decimal ROUND at scales 401 and -401, while our documented native domain rejects them.

This uses the existing ValidatorApi.doExprValidate path, not an additional full validation pass. A rejected expression becomes a failed offload candidate through the normal validation/fallback path. I am retaining the guard to preserve this behavior independently of the optional native-validation flag.

GlutenDecimalRoundSuite explicitly exercises this configuration and checks both the fallback plan and Spark-compatible results:

testGluten(
"decimal ROUND retains NULL scales and guards extreme scales without native validation") {
withInput(DecimalType(3, 1), Seq("2.5", "-2.5", "0")) {
withSQLConf(
GlutenConfig.NATIVE_VALIDATION_ENABLED.key -> "false",
SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> NullPropagation.ruleName) {
checkRound("CAST(NULL AS INT)")
checkRound("-401", native = false)
checkRound("401", native = false)
}
}
}
testGluten("decimal ROUND accepts foldable scales with constant folding disabled") {
withInput(DecimalType(3, 1), Seq("2.5", "-2.5", "0")) {
for (nativeValidation <- Seq("false", "true")) {
withSQLConf(
GlutenConfig.NATIVE_VALIDATION_ENABLED.key -> nativeValidation,
SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
s"${ConstantFolding.ruleName},${NullPropagation.ruleName}"
) {
val scales = sql("SELECT round(value, 1 - 2) FROM decimal_round_input")
.queryExecution.optimizedPlan.collect {
case node => node.expressions.flatMap(_.collect { case round: Round => round.scale })
}.flatten
assert(scales.nonEmpty)
assert(scales.forall(scale => scale.foldable && !scale.isInstanceOf[Literal]))
checkRound("1 - 2")
checkRound("-200 - 200")
checkRound("200 + 200")
checkRound("CAST(NULL AS INT)")
checkRound("400 + 1", native = false)
checkRound("-400 - 1", native = false)
}
}
}
}
}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

spark.gluten.sql.enable.native.validation = false

The intention is to perform native validation whenever possible, while avoiding introducing native-engine-specific logic into Scala. Instead, we aim to maintain a unified interface between Scala and native validation.

This configuration is primarily intended for debugging. It forces all computation to be offloaded to the native engine without fallback, meaning that unsupported operations will result in runtime exceptions rather than falling back to the original engine.

Does this approach make sense to you? Feel free to update the relevant documentation if anything is unclear.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@rui-mo, agreed. I removed the decimal-specific doExprValidate policy from Scala and now rely on the existing native Substrait validator as the normal compatibility gate.

Commit: bb36d43

The behavior is now:

  • With native validation enabled (the default), scales outside [-400, 400] fail native validation and the expression falls back to Spark.
  • With native validation disabled for debugging, the plan remains native and the registered decimal special form reports its construction-time scale error at execution.

I also updated the internal configuration description to make that debug-mode behavior explicit.

The direct validator regression now verifies that Scala does not filter or eagerly evaluate invalid literals, nonliteral scales, foldable expressions, or an ANSI-true throwing cast. Integration tests cover normal fallback versus debug-mode native failure, while typed NULL and bounded foldable scales continue to execute natively.

Validation passed: native 3/3, API 2/2, Spark 4.1 SQL 4/4 on JDK 17 and 21, full Spark 4.1 reactor build, active Spotless/Scalastyle, license headers, and packaged native-library hash checks. Fresh upstream CI is still required.

val scale = round.scale
val supported =
if (scale.dataType != IntegerType || !scale.foldable) {
false
} else {
scale.eval(EmptyRow) match {
case null => true
case value: Int => value >= -400 && value <= 400
case _ => false
}
}
if (!supported) {
logDebug("Decimal ROUND requires a foldable INTEGER scale in [-400, 400] or NULL.")
}
supported
case _ => true
}

override def doNativeValidateWithFailureReason(plan: PlanNode): ValidationResult = {
TaskResources.runUnsafe {
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
package org.apache.gluten.backendsapi.velox

import org.apache.gluten.config.GlutenConfig
import org.apache.gluten.expression.{DecimalRoundTransformer, LiteralTransformer}
import org.apache.gluten.substrait.SubstraitContext

import org.apache.spark.sql.catalyst.expressions.{Add, BoundReference, Cast, Literal, Round}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._

import org.scalatest.funsuite.AnyFunSuite

class VeloxDecimalRoundSuite extends AnyFunSuite {
test("decimal round preserves a typed NULL scale") {
val value = Literal.create(Decimal(new java.math.BigDecimal("2.5")), DecimalType(3, 1))
val scale = Literal.create(null, IntegerType)
val original = Round(value, scale)
val transformer = DecimalRoundTransformer("round", LiteralTransformer(value), original)
val scalar = transformer.doTransform(new SubstraitContext).toProtobuf.getScalarFunction
assert(scalar.getArgumentsCount == 2)
assert(scalar.getArguments(1).getValue.getLiteral.hasNull)
assert(transformer.dataType == DecimalType(3, 0))
}

test("decimal round scale guards remain active without general native validation") {
val conf = new SQLConf
conf.setConfString(GlutenConfig.NATIVE_VALIDATION_ENABLED.key, "false")
SQLConf.withExistingConf(conf) {
val validator = new VeloxValidatorApi
val value = BoundReference(0, DecimalType(38, 38), nullable = true)
Seq(-400, -39, -1, 0, 38, 400).foreach {
scale => assert(validator.doExprValidate("round", Round(value, Literal(scale))))
}
Seq(Int.MinValue, -401, 401, Int.MaxValue).foreach {
scale => assert(!validator.doExprValidate("round", Round(value, Literal(scale))))
}
assert(validator.doExprValidate("round", Round(value, Literal.create(null, IntegerType))))
assert(!validator.doExprValidate(
"round",
Round(value, BoundReference(1, IntegerType, nullable = false))))
}
}

test("decimal round accepts bounded foldable INTEGER scales without prior constant folding") {
val conf = new SQLConf
conf.setConfString(GlutenConfig.NATIVE_VALIDATION_ENABLED.key, "false")
conf.setConfString(SQLConf.ANSI_ENABLED.key, "true")
SQLConf.withExistingConf(conf) {
val validator = new VeloxValidatorApi
val value = BoundReference(0, DecimalType(38, 38), nullable = true)
Seq(-400, -1, 0, 400).foreach {
scale =>
assert(validator.doExprValidate(
"round",
Round(value, Add(Literal(scale), Literal(0)))))
}
assert(validator.doExprValidate(
"round",
Round(value, Cast(Literal.create(null, NullType), IntegerType))))
Seq(Add(Literal(-400), Literal(-1)), Add(Literal(400), Literal(1))).foreach {
scale => assert(!validator.doExprValidate("round", Round(value, scale)))
}
assert(!validator.doExprValidate(
"round",
Round(value, Cast(Literal.create(null, NullType), LongType))))
val error = intercept[Exception] {
validator.doExprValidate("round", Round(value, Cast(Literal("invalid-scale"), IntegerType)))
}
assert(error.getMessage.contains("invalid-scale"))
}
}
}
17 changes: 15 additions & 2 deletions cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,7 @@ bool SubstraitToVeloxPlanValidator::flattenDualLevel(const TypePtr& type, std::v

bool SubstraitToVeloxPlanValidator::validateRound(
const ::substrait::Expression::ScalarFunction& scalarFunction,
const RowTypePtr& inputType) {
const std::vector<core::TypedExprPtr>& params) {
const auto& arguments = scalarFunction.arguments();
if (arguments.size() < 2) {
return false;
Expand All @@ -127,6 +127,19 @@ bool SubstraitToVeloxPlanValidator::validateRound(
return false;
}

const auto& valueType = params[0]->type();
if (valueType->isDecimal()) {
const auto& scale = arguments[1].value().literal();
if (scale.has_null() && scale.null().has_i32()) {
return true;
}
if (scale.has_i32() && scale.i32() >= -400 && scale.i32() <= 400) {
return true;
}
LOG_VALIDATION_MSG("Decimal ROUND requires an INTEGER scale in [-400, 400] or NULL.");
return false;
}
Comment thread
luis4a0 marked this conversation as resolved.
Resolved

// Velox has different result with Spark on negative scale.
auto typeCase = arguments[1].value().literal().literal_type_case();
switch (typeCase) {
Expand Down Expand Up @@ -213,7 +226,7 @@ bool SubstraitToVeloxPlanValidator::validateScalarFunction(
std::vector<std::string> types = SubstraitParser::getSubFunctionTypes(function);

if (name == "round") {
return validateRound(scalarFunction, inputType);
return validateRound(scalarFunction, params);
}
if (name == "extract") {
return validateExtractExpr(params);
Expand Down
7 changes: 5 additions & 2 deletions cpp/velox/substrait/SubstraitToVeloxPlanValidator.h
Original file line number Diff line number Diff line change
Expand Up @@ -148,8 +148,11 @@ class SubstraitToVeloxPlanValidator {
/// Validate aggregate rel.
bool validateAggRelFunctionType(const ::substrait::AggregateRel& substraitAgg);

/// Validate the round scalar function.
bool validateRound(const ::substrait::Expression::ScalarFunction& scalarFunction, const RowTypePtr& inputType);
/// Validate the round scalar function using its already-converted arguments.
/// The caller supplies one non-null expression for each Substrait argument.
Comment on lines +152 to +153
bool validateRound(
const ::substrait::Expression::ScalarFunction& scalarFunction,
const std::vector<core::TypedExprPtr>& params);

/// Validate extract function.
bool validateExtractExpr(const std::vector<core::TypedExprPtr>& params);
Expand Down
45 changes: 45 additions & 0 deletions cpp/velox/tests/SparkFunctionTest.cc
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,13 @@
* limitations under the License.
*/

#include <limits>
#include <optional>
#include <string>
#include <vector>

#include "operators/functions/RegistrationAllFunctions.h"
#include "substrait/SubstraitToVeloxPlanValidator.h"
#include "velox/common/base/tests/GTestUtils.h"
#include "velox/core/Expressions.h"
#include "velox/functions/sparksql/SparkQueryConfig.h"
Expand Down Expand Up @@ -125,6 +128,48 @@ TEST_F(SparkFunctionTest, roundWithDecimal) {
runRoundWithDecimalTest<int8_t>(testRoundWithDecIntegralData<int8_t>());
}

TEST_F(SparkFunctionTest, decimalRoundScaleValidation) {
const auto validate = [&](std::optional<int32_t> scale, bool castInput) {
::substrait::Expression expression;
auto* function = expression.mutable_scalar_function();
function->set_function_reference(1);
auto* value = function->add_arguments()->mutable_value();
if (castInput) {
auto* cast = value->mutable_cast();
auto* type = cast->mutable_type()->mutable_decimal();
type->set_precision(38);
type->set_scale(0);
type->set_nullability(::substrait::Type_Nullability_NULLABILITY_NULLABLE);
value = cast->mutable_input();
}
value->mutable_selection()->mutable_direct_reference()->mutable_struct_field()->set_field(0);
auto* literal = function->add_arguments()->mutable_value()->mutable_literal();
if (scale.has_value()) {
literal->set_i32(scale.value());
} else {
literal->mutable_null()->mutable_i32()->set_nullability(::substrait::Type_Nullability_NULLABILITY_NULLABLE);
}
auto* output = function->mutable_output_type()->mutable_decimal();
output->set_precision(38);
output->set_scale(0);
output->set_nullability(::substrait::Type_Nullability_NULLABILITY_NULLABLE);
gluten::SubstraitToVeloxPlanValidator validator(pool());
return validator.validate(
expression, ROW({"value"}, {castInput ? VARCHAR() : DECIMAL(38, 0)}), {{1, "round:opt_dec<38,0>_i32"}});
};
for (const auto scale : {-400, -39, -38, -1, 0, 38, 400}) {
SCOPED_TRACE(scale);
EXPECT_TRUE(validate(scale, false));
}
EXPECT_TRUE(validate(std::nullopt, false));
EXPECT_TRUE(validate(-1, true));
EXPECT_TRUE(validate(std::nullopt, true));
for (const auto scale : {std::numeric_limits<int32_t>::min(), -401, 401, std::numeric_limits<int32_t>::max()}) {
SCOPED_TRACE(scale);
EXPECT_FALSE(validate(scale, false));
}
}

TEST_F(SparkFunctionTest, expressionLevelAnsiCastIgnoresSessionAnsiOff) {
queryCtx_->testingOverrideConfigUnsafe({{sparkAnsiEnabledConfigKey(), "false"}});
auto input = makeRowVector({makeFlatVector<std::string>({"2147483648"})});
Expand Down
3 changes: 1 addition & 2 deletions docs/velox-backend-scalar-function-support.md
Original file line number Diff line number Diff line change
Expand Up @@ -297,7 +297,7 @@ The status applies to `spark.sql.ansi.enabled=false`. When ANSI mode is enabled,
| randn | Randn | S | |
| random | Rand | S | |
| rint | Rint | S | |
| round | Round | S | |
| round | Round | PS | For decimal inputs, foldable INTEGER scales in [-400, 400] and NULL are supported. Other decimal scales fall back to Spark. Integral and floating-point negative scales retain their existing fallback. |
| sec | Sec | S | |
| sign | Signum | S | |
| signum | Signum | S | |
Expand Down Expand Up @@ -538,4 +538,3 @@ The status applies to `spark.sql.ansi.enabled=false`. When ANSI mode is enabled,
| xpath_number | XPathDouble | | |
| xpath_short | XPathShort | | |
| xpath_string | XPathString | | |

2 changes: 1 addition & 1 deletion ep/build-velox/src/get-velox.sh
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ RUN_SETUP_SCRIPT=ON
ENABLE_ENHANCED_FEATURES=OFF

# Developer use only for testing Velox PR.
UPSTREAM_VELOX_PR_ID=""
UPSTREAM_VELOX_PR_ID="19141"

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed that this must not become the normal merged dependency. This revision is intentionally a dependency-blocked validation companion: Rui explicitly requested exercising the upstream PR through the existing developer hook at facebookincubator/velox#19141 (comment) . Clearing the hook now would test the stock dependency without the decimal fix. The PR description records the merge gate: land the upstream fix, qualify a normal dependency containing it, then restore the empty hook before merging. Keeping this thread open to track that required dependency step.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@luis4a0 Yes, this is only for development testing, and we could remove this change after the corresponding Velox gets merged.

Comment thread
luis4a0 marked this conversation as resolved.
Resolved
Comment thread
luis4a0 marked this conversation as resolved.
Resolved

OS=`uname -s`

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,15 +20,16 @@ import org.apache.gluten.backendsapi.BackendsApiManager
import org.apache.gluten.exception.GlutenNotSupportException

import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.types.{DataType, DecimalType}
import org.apache.spark.sql.types.{DataType, DecimalType, IntegerType}

case class DecimalRoundTransformer(
substraitExprName: String,
child: ExpressionTransformer,
original: Round)
extends BinaryExpressionTransformer {

val toScale: Int = original.scale.eval(EmptyRow).asInstanceOf[Int]
private val scaleValue = original.scale.eval(EmptyRow)
val toScale: Int = if (scaleValue == null) 0 else scaleValue.asInstanceOf[Int]

// Use the same result type for different Spark versions in velox.
// The same result type with spark in ch.
Expand All @@ -43,5 +44,6 @@ case class DecimalRoundTransformer(
}

override def left: ExpressionTransformer = child
override def right: ExpressionTransformer = LiteralTransformer(toScale)
override def right: ExpressionTransformer =
LiteralTransformer(Literal.create(scaleValue, IntegerType))
}
Loading
Loading