[VL] Offload 2-arg decimal ceiling/floor to Velox native execution

backends-velox/src/test/scala/org/apache/gluten/functions/MathFunctionsValidateSuite.scala

@@ -122,6 +122,30 @@ 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.
+    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("cos") {
     runQueryAndCompare("SELECT cos(l_orderkey) from lineitem limit 1") {
       checkGlutenPlan[ProjectExecTransformer]

cpp/velox/substrait/SubstraitParser.cc

@@ -254,10 +254,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" ||
@@ -267,6 +267,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;

cpp/velox/substrait/SubstraitParser.h

@@ -80,7 +80,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.

gluten-substrait/src/main/scala/org/apache/gluten/expression/DecimalCeilFloorTransformer.scala

@@ -0,0 +1,63 @@
+/*
+ * 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 {
+
+  private val toScale: Int = {
+    val evaluated = scaleExpr.eval(EmptyRow)
+    if (evaluated == null) {
+      throw new GlutenNotSupportException(
+        s"Scale expression evaluated to null for ${original.nodeName}. Falling back to Spark.")
+    }
+    evaluated.asInstanceOf[Int]
+  }
+
+  override val dataType: DataType = original.children.head.dataType match {
+    case decimalType: DecimalType =>
+      BackendsApiManager.getSparkPlanExecApiInstance.genDecimalRoundExpressionOutput(
+        decimalType,
+        toScale)
+    case other =>
+      throw new GlutenNotSupportException(
+        s"Decimal type is expected for ${original.nodeName} but received ${other.typeName}.")
+  }
+
+  override def left: ExpressionTransformer = child
+  override def right: ExpressionTransformer = LiteralTransformer(toScale)
+}

gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala

@@ -394,6 +394,18 @@ object ExpressionConverter extends SQLConfHelper with Logging {
           substraitExprName,
           replaceWithExpressionTransformer0(r.child, attributeSeq, expressionsMap),
           r)
+      case rc: RoundCeil if rc.child.dataType.isInstanceOf[DecimalType] =>
+        DecimalCeilFloorTransformer(
+          substraitExprName,
+          replaceWithExpressionTransformer0(rc.child, attributeSeq, expressionsMap),
+          rc,
+          rc.scale)
+      case rf: RoundFloor if rf.child.dataType.isInstanceOf[DecimalType] =>
+        DecimalCeilFloorTransformer(
+          substraitExprName,
+          replaceWithExpressionTransformer0(rf.child, attributeSeq, expressionsMap),
+          rf,
+          rf.scale)
       case t: ToUnixTimestamp =>
         BackendsApiManager.getSparkPlanExecApiInstance.genToUnixTimestampTransformer(
           substraitExprName,