This is an automated email from the ASF dual-hosted git repository.
weiting-chen pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new aaec0f5161 [VL] Offload 2-arg decimal ceiling/floor to Velox native
execution (#12777)
aaec0f5161 is described below
commit aaec0f51615a0a573ad3484bb21ab1b6331eb400
Author: Minni Mittal <[email protected]>
AuthorDate: Fri Oct 30 19:43:51 2026 +0530
[VL] Offload 2-arg decimal ceiling/floor to Velox native execution (#12777)
* [VL] Offload 2-arg decimal ceiling/floor to Velox native execution
Spark's `ceiling(x, scale)` / `floor(x, scale)` on decimal inputs produce
`RoundCeil(decimal, scale)` / `RoundFloor(decimal, scale)`. These were
mapped
to the substrait `ceil` / `floor` function names but had no native execution
path, so the 2-arg decimal form fell back to vanilla Spark.
This wires them to the Velox `decimal_ceil` / `decimal_floor` special forms:
- SubstraitParser: `mapToVeloxFunction` gains a `numArgs` parameter and
remaps
2-arg `ceil` / `floor` on decimals to `decimal_ceil` / `decimal_floor`.
Unary
`ceil(decimal)` / `floor(decimal)` keep their existing name.
- DecimalCeilFloorTransformer: new transformer mirroring
DecimalRoundTransformer;
recomputes the output DecimalType from the input type and constant-folded
scale
(matching Spark's RoundBase.dataType) and emits the scale as a literal.
- ExpressionConverter: route decimal RoundCeil / RoundFloor through the new
transformer.
Addresses the RoundCeil / RoundFloor items in #10134.
* [VL] Address review: ANSI fallback + defensive scale handling for 2-arg
decimal ceil/floor
Address review feedback on the 2-arg decimal ceiling/floor offload:
- Overflow semantics: Velox's decimal_ceil / decimal_floor return NULL when
the rounded result exceeds the declared precision, whereas Spark's
RoundBase
raises under ANSI mode. DecimalCeilFloorTransformer now falls back to
Spark
when ANSI mode is enabled, so the ANSI error is preserved instead of a
silent
NULL (this holds even when spark.gluten.sql.ansiFallback.enabled is
false).
Under non-ANSI mode Spark also returns NULL, so native offload stays
correct.
- Defensive scale handling: require a foldable scale and wrap evaluation /
type
errors in GlutenNotSupportException so a non-literal or un-evaluable scale
triggers a clean Spark fallback instead of aborting transformation.
- Tests: gate the 2-arg ceiling/floor test with SparkVersionUtil.gteSpark40
(the
SQL form is invalid on Spark 3.4/3.5 and would fail during analysis), and
add
a regression test asserting the ANSI overflow case falls back and raises
rather
than returning NULL.
Generated-by: GitHub Copilot CLI (Claude Opus 4.8)
Co-authored-by: Copilot <[email protected]>
* [VL] Preserve decimal ceil/floor overflow semantics
## Summary
Preserve Spark behavior when offloading two-argument decimal ceiling and
floor expressions. Fall back for precision-overflow and unsupported
negative-scale cases while retaining native execution for safe inputs, with
focused cross-version regression coverage.
## Prompting Intent
Address all open review comments on apache/gluten#12777, validate the
changes through the six-round review gauntlet, keep unsupported cases on Spark,
and ensure the public commit contains no internal or Microsoft-only material.
## Linked Sources
- Pull request: https://github.com/apache/gluten/pull/12777
- Tracking issue: https://github.com/apache/gluten/issues/10134
- Review comments:
https://github.com/apache/gluten/pull/12777#pullrequestreview-threads
## Rationale
Velox returns NULL for decimal directional-rounding overflow while Spark
raises regardless of ANSI mode. Type-level eligibility checks avoid silent
semantic divergence without disabling native execution for safe precision and
scale combinations. Negative-scale Spark decimals fall back because Velox
decimal types cannot represent them safely.
Co-authored-by: Copilot <[email protected]>
Copilot-Session: 8f9d9624-2e39-43dd-b723-f541d372a92c
* [VL] Scope decimal ceil/floor conversion to Velox
## Summary
Move the decimal RoundCeil and RoundFloor transformer selection from the
shared expression converter into the Velox backend-specific converter hook.
Bolt and ClickHouse retain their existing generic conversion and fallback
behavior.
## Prompting Intent
Address the new review finding on apache/gluten#12777 that the shared
converter exposed Velox-only decimal ceiling and floor semantics to the Bolt
backend, while keeping the contribution public and free of internal material.
## Linked Sources
- Pull request: https://github.com/apache/gluten/pull/12777
- Review thread:
https://github.com/apache/gluten/pull/12777#discussion_r4141110964
- Tracking issue: https://github.com/apache/gluten/issues/10134
## Rationale
Used the existing backend-specific expression converter hook instead of
adding backend-name checks to common code. This keeps the common module
backend-neutral and confines the specialized decimal_ceil and decimal_floor
path to the only backend that implements it.
Co-authored-by: Copilot <[email protected]>
Copilot-Session: 8f9d9624-2e39-43dd-b723-f541d372a92c
* [VL] Format decimal ceil/floor tests
## Summary
Apply the repository Scala formatter to the decimal ceiling and floor
regression tests so the Scala format CI check passes without changing behavior.
## Prompting Intent
Investigate and fix the Scala formatting failure reported by CI for
apache/gluten#12777 while keeping the public contribution free of internal
material.
## Linked Sources
- Pull request: https://github.com/apache/gluten/pull/12777
- Failed Scala format run:
https://github.com/apache/gluten/actions/runs/36673625270
## Rationale
Used the repository-provided Scala formatter rather than manually
approximating its layout rules. The formatter modified only the reported test
file, and a subsequent format check completed successfully.
Co-authored-by: Copilot <[email protected]>
Copilot-Session: 8f9d9624-2e39-43dd-b723-f541d372a92c
---------
Co-authored-by: Copilot <[email protected]>
Copilot-Session: 8f9d9624-2e39-43dd-b723-f541d372a92c
---
.../backendsapi/velox/VeloxSparkPlanExecApi.scala | 23 ++++
.../functions/MathFunctionsValidateSuite.scala | 126 +++++++++++++++++++++
cpp/velox/substrait/SubstraitParser.cc | 11 +-
cpp/velox/substrait/SubstraitParser.h | 7 +-
.../expression/DecimalCeilFloorTransformer.scala | 111 ++++++++++++++++++
5 files changed, 275 insertions(+), 3 deletions(-)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
index 4c64695507..4a7c2328d4 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
@@ -1405,6 +1405,29 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi
with Logging {
)
}
+ override def extraExpressionConverter(
+ substraitExprName: String,
+ expr: Expression,
+ attributeSeq: Seq[Attribute]): Option[ExpressionTransformer] = {
+ expr match {
+ case rc: RoundCeil if rc.child.dataType.isInstanceOf[DecimalType] =>
+ Some(
+ DecimalCeilFloorTransformer(
+ substraitExprName,
+ ExpressionConverter.replaceWithExpressionTransformer(rc.child,
attributeSeq),
+ rc,
+ rc.scale))
+ case rf: RoundFloor if rf.child.dataType.isInstanceOf[DecimalType] =>
+ Some(
+ DecimalCeilFloorTransformer(
+ substraitExprName,
+ ExpressionConverter.replaceWithExpressionTransformer(rf.child,
attributeSeq),
+ rf,
+ rf.scale))
+ case _ => None
+ }
+ }
+
override def rewriteSpillPath(path: String): String = {
val fs = VeloxConfig.get.veloxSpillFileSystem
fs match {
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/functions/MathFunctionsValidateSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/functions/MathFunctionsValidateSuite.scala
index 13f3f4feea..5d64b5fd33 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/functions/MathFunctionsValidateSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/functions/MathFunctionsValidateSuite.scala
@@ -21,7 +21,10 @@ import
org.apache.gluten.execution.{BatchScanExecTransformer, ProjectExecTransfo
import org.apache.spark.SparkConf
import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.expressions.{RoundCeil, RoundFloor}
+import org.apache.spark.sql.functions.col
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.DecimalType
class MathFunctionsValidateSuiteAnsiOn extends FunctionsValidateSuite {
@@ -122,6 +125,129 @@ class MathFunctionsValidateSuite extends
FunctionsValidateSuite {
}
}
+ test("2-arg ceiling / floor on decimals (RoundCeil / RoundFloor)") {
+ // The 2-arg forms produce Spark RoundCeil / RoundFloor and dispatch to
the Velox
+ // decimal_ceil / decimal_floor special forms. The projection is native
only when the
+ // expression offloads, so checkGlutenPlan[ProjectExecTransformer] doubles
as an offload
+ // assertion; runQueryAndCompare additionally validates results against
vanilla Spark.
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+ runQueryAndCompare(
+ "SELECT ceiling(cast(l_quantity as decimal(12, 2)), 1) FROM lineitem
limit 10") {
+ checkGlutenPlan[ProjectExecTransformer]
+ }
+ runQueryAndCompare(
+ "SELECT floor(cast(l_quantity as decimal(12, 2)), 1) FROM lineitem
limit 10") {
+ checkGlutenPlan[ProjectExecTransformer]
+ }
+ // Negative scale rounds to the left of the decimal point.
+ runQueryAndCompare(
+ "SELECT ceiling(cast(l_extendedprice as decimal(20, 4)), -2) FROM
lineitem limit 10") {
+ checkGlutenPlan[ProjectExecTransformer]
+ }
+ runQueryAndCompare(
+ "SELECT floor(cast(l_extendedprice as decimal(20, 4)), -2) FROM
lineitem limit 10") {
+ checkGlutenPlan[ProjectExecTransformer]
+ }
+ }
+ }
+
+ test("2-arg ceiling / floor on decimals offloads under ANSI when precision
cannot overflow") {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ GlutenConfig.GLUTEN_ANSI_FALLBACK_ENABLED.key -> "false") {
+ runQueryAndCompare(
+ "SELECT ceiling(cast(l_quantity as decimal(12, 2)), 1) FROM lineitem
limit 10") {
+ checkGlutenPlan[ProjectExecTransformer]
+ }
+ runQueryAndCompare(
+ "SELECT floor(cast(l_extendedprice as decimal(20, 4)), -2) FROM
lineitem limit 10") {
+ checkGlutenPlan[ProjectExecTransformer]
+ }
+ }
+ }
+
+ test("2-arg ceiling / floor on decimals falls back when precision can
overflow") {
+ Seq("true", "false").foreach {
+ ansiEnabled =>
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> ansiEnabled,
+ GlutenConfig.GLUTEN_ANSI_FALLBACK_ENABLED.key -> "false") {
+ Seq(
+ ("ceiling", "99999999999999999999999999999999999999", "decimal(38,
0)", "-1"),
+ ("floor", "-99999999999999999999999999999999999999", "decimal(38,
0)", "-1"),
+ ("ceiling", "5", "decimal(10, 2)", "-38")
+ ).foreach {
+ case (function, value, decimalType, scale) =>
+ val df =
+ spark.sql(s"SELECT $function(cast('$value' as $decimalType),
$scale)")
+ assert(
+ df.queryExecution.executedPlan
+ .collect { case project: ProjectExecTransformer => project }
+ .isEmpty,
+ s"Expected decimal $function overflow to fall back to
Spark:\n" +
+ df.queryExecution.executedPlan
+ )
+ val error = intercept[Exception] {
+ df.collect()
+ }
+ assert(
+ Iterator
+ .iterate[Throwable](error)(_.getCause)
+ .takeWhile(_ != null)
+ .exists(_.isInstanceOf[ArithmeticException]),
+ s"Expected an arithmetic overflow error but received: $error"
+ )
+ }
+
+ val df =
+ spark.sql(
+ "SELECT ceiling(cast('0' as decimal(10, 2)), cast(-2147483648 as
int))")
+ assert(
+ df.queryExecution.executedPlan
+ .collect { case project: ProjectExecTransformer => project }
+ .isEmpty,
+ s"Expected extreme-scale decimal ceiling to fall back to Spark:\n"
+
+ df.queryExecution.executedPlan
+ )
+ assert(df.collect().head.getDecimal(0).signum() == 0)
+ }
+ }
+ }
+
+ test("2-arg ceiling / floor on negative-scale decimals falls back") {
+ withSQLConf(
+ SQLConf.LEGACY_ALLOW_NEGATIVE_SCALE_OF_DECIMAL_ENABLED.key -> "true",
+ SQLConf.ANSI_ENABLED.key -> "true",
+ GlutenConfig.GLUTEN_ANSI_FALLBACK_ENABLED.key -> "false"
+ ) {
+ val input = spark
+ .range(-1L, 2L)
+ .select(col("id").cast(DecimalType(1, -256)).as("d"))
+
+ Seq("ceiling(d, -255)", "floor(d, -255)").foreach {
+ expression =>
+ val df = input.selectExpr(expression)
+ assert(
+ df.queryExecution.optimizedPlan.expressions.exists(
+ _.find {
+ case _: RoundCeil | _: RoundFloor => true
+ case _ => false
+ }.isDefined),
+ s"Expected $expression to remain in the optimized plan:\n" +
+ df.queryExecution.optimizedPlan
+ )
+ assert(
+ df.queryExecution.executedPlan
+ .collect { case project: ProjectExecTransformer => project }
+ .isEmpty,
+ s"Expected $expression on a negative-scale decimal to fall back to
Spark:\n" +
+ df.queryExecution.executedPlan
+ )
+ assert(df.collect().forall(_.getDecimal(0).signum() == 0))
+ }
+ }
+ }
+
test("cos") {
runQueryAndCompare("SELECT cos(l_orderkey) from lineitem limit 1") {
checkGlutenPlan[ProjectExecTransformer]
diff --git a/cpp/velox/substrait/SubstraitParser.cc
b/cpp/velox/substrait/SubstraitParser.cc
index 869c42b62c..3333810e01 100644
--- a/cpp/velox/substrait/SubstraitParser.cc
+++ b/cpp/velox/substrait/SubstraitParser.cc
@@ -274,10 +274,10 @@ std::string SubstraitParser::findVeloxFunction(
break;
}
}
- return mapToVeloxFunction(funcName, isDecimal);
+ return mapToVeloxFunction(funcName, isDecimal, types.size());
}
-std::string SubstraitParser::mapToVeloxFunction(const std::string&
substraitFunction, bool isDecimal) {
+std::string SubstraitParser::mapToVeloxFunction(const std::string&
substraitFunction, bool isDecimal, size_t numArgs) {
auto it = substraitVeloxFunctionMap_.find(substraitFunction);
if (isDecimal) {
if (substraitFunction == "lt" || substraitFunction == "lte" ||
substraitFunction == "gt" ||
@@ -287,6 +287,13 @@ std::string SubstraitParser::mapToVeloxFunction(const
std::string& substraitFunc
if (substraitFunction == "round") {
return "decimal_round";
}
+ // Spark RoundCeil / RoundFloor are emitted with substrait names "ceil"
+ // and "floor" but require dispatch to the 2-arg decimal special forms.
+ // The unary forms `ceil(decimal)` / `floor(decimal)` keep their original
+ // name (handled by simple-function registration).
+ if (numArgs == 2 && (substraitFunction == "ceil" || substraitFunction ==
"floor")) {
+ return "decimal_" + substraitFunction;
+ }
}
if (it != substraitVeloxFunctionMap_.end()) {
return it->second;
diff --git a/cpp/velox/substrait/SubstraitParser.h
b/cpp/velox/substrait/SubstraitParser.h
index 8587c1e5f4..8986f17232 100644
--- a/cpp/velox/substrait/SubstraitParser.h
+++ b/cpp/velox/substrait/SubstraitParser.h
@@ -86,7 +86,12 @@ class SubstraitParser {
static std::string findVeloxFunction(const std::unordered_map<uint64_t,
std::string>& functionMap, uint64_t id);
/// Map the Substrait function keyword into Velox function keyword.
- static std::string mapToVeloxFunction(const std::string& substraitFunction,
bool isDecimal);
+ /// `numArgs` is the number of substrait arguments observed in the call
+ /// signature. It is used to disambiguate overloads such as unary
+ /// `ceil(decimal)` (mapped to `ceil`) vs. 2-arg `ceil(decimal, int)`
+ /// (mapped to the special form `decimal_ceil`). Pass 0 when arity is
+ /// unknown -- in that case arity-sensitive remappings are skipped.
+ static std::string mapToVeloxFunction(const std::string& substraitFunction,
bool isDecimal, size_t numArgs = 0);
/// @brief Return whether a config is set as true in AdvancedExtension
/// optimization.
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/expression/DecimalCeilFloorTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/expression/DecimalCeilFloorTransformer.scala
new file mode 100644
index 0000000000..477a08f23a
--- /dev/null
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/expression/DecimalCeilFloorTransformer.scala
@@ -0,0 +1,111 @@
+/*
+ * 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.expression
+
+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}
+
+/**
+ * Transformer for Spark `RoundCeil(decimal, scale)` and `RoundFloor(decimal,
scale)`. These power
+ * the 2-argument forms of `ceiling(x, scale)` / `floor(x, scale)` and
dispatch to the Velox
+ * `decimal_ceil` / `decimal_floor` special forms (substrait names `ceil` /
`floor`, remapped on the
+ * C++ side based on arity + decimal arg type).
+ *
+ * The output `DataType` is recomputed from the original Spark decimal input
type and the constant
+ * folded scale, matching Spark's `RoundBase.dataType` formula. Mirrors the
structure of
+ * `DecimalRoundTransformer`.
+ */
+case class DecimalCeilFloorTransformer(
+ substraitExprName: String,
+ child: ExpressionTransformer,
+ original: Expression,
+ scaleExpr: Expression)
+ extends BinaryExpressionTransformer {
+
+ // Spark requires the scale to be a foldable integer literal, but guard
defensively so any
+ // non-foldable scale, evaluation failure, or unexpected value type triggers
a clean fallback
+ // via GlutenNotSupportException instead of aborting the whole
transformation.
+ private val toScale: Int = {
+ if (!scaleExpr.foldable) {
+ throw new GlutenNotSupportException(
+ s"Scale expression is not foldable for ${original.nodeName}. Falling
back to Spark.")
+ }
+ val evaluated =
+ try {
+ scaleExpr.eval(EmptyRow)
+ } catch {
+ case e: Exception =>
+ throw new GlutenNotSupportException(
+ s"Failed to evaluate scale expression for ${original.nodeName}:
${e.getMessage}. " +
+ "Falling back to Spark.")
+ }
+ evaluated match {
+ case null =>
+ throw new GlutenNotSupportException(
+ s"Scale expression evaluated to null for ${original.nodeName}.
Falling back to Spark.")
+ case i: Int => i
+ case other =>
+ throw new GlutenNotSupportException(
+ s"Scale expression for ${original.nodeName} is expected to be an int
but evaluated to " +
+ s"${other.getClass.getSimpleName}. Falling back to Spark.")
+ }
+ }
+
+ private val decimalType = original.children.head.dataType match {
+ case decimalType: DecimalType =>
+ decimalType
+ case other =>
+ throw new GlutenNotSupportException(
+ s"Decimal type is expected for ${original.nodeName} but received
${other.typeName}.")
+ }
+
+ if (decimalType.scale < 0) {
+ throw new GlutenNotSupportException(
+ s"${original.nodeName} does not support native execution for
${decimalType.sql} because " +
+ "Velox decimal types require a non-negative scale. Falling back to
Spark.")
+ }
+
+ private val requiredPrecision: Long = {
+ val integralLeastNumDigits = decimalType.precision.toLong -
decimalType.scale + 1L
+ if (toScale < 0) {
+ math.max(integralLeastNumDigits, -toScale.toLong + 1L)
+ } else {
+ integralLeastNumDigits + math.min(decimalType.scale, toScale)
+ }
+ }
+
+ // Velox returns NULL on decimal precision overflow, while Spark raises
regardless of ANSI mode.
+ // Spark caps the output type at MAX_PRECISION, so fall back only when
rounding can change the
+ // value and the uncapped result precision exceeds that limit.
+ if (toScale < decimalType.scale && requiredPrecision >
DecimalType.MAX_PRECISION) {
+ throw new GlutenNotSupportException(
+ s"${original.nodeName} on ${decimalType.sql} with scale $toScale may
exceed decimal " +
+ s"precision ${DecimalType.MAX_PRECISION} (required precision:
$requiredPrecision). " +
+ "Falling back to Spark.")
+ }
+
+ override val dataType: DataType =
+
BackendsApiManager.getSparkPlanExecApiInstance.genDecimalRoundExpressionOutput(
+ decimalType,
+ toScale)
+
+ override def left: ExpressionTransformer = child
+ override def right: ExpressionTransformer = LiteralTransformer(toScale)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
