This is an automated email from the ASF dual-hosted git repository.
agrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new acfd03cab feat:support ansi mode rounding function (#2542)
acfd03cab is described below
commit acfd03cabc806c2bf73aac7f3f3f634e2954a6f1
Author: B Vadlamani <[email protected]>
AuthorDate: Mon Oct 13 16:21:56 2025 -0700
feat:support ansi mode rounding function (#2542)
---
docs/source/user-guide/latest/expressions.md | 4 +-
native/core/src/execution/planner.rs | 4 +-
native/proto/src/proto/expr.proto | 1 +
native/spark-expr/src/comet_scalar_funcs.rs | 4 +-
native/spark-expr/src/math_funcs/round.rs | 87 ++++++++++++++++------
.../org/apache/comet/serde/QueryPlanSerde.scala | 6 +-
.../scala/org/apache/comet/serde/arithmetic.scala | 15 ++--
.../main/scala/org/apache/comet/serde/arrays.scala | 2 +
.../scala/org/apache/comet/serde/bitwise.scala | 4 +-
.../scala/org/apache/comet/serde/datetime.scala | 7 +-
.../main/scala/org/apache/comet/serde/hash.scala | 10 ++-
.../main/scala/org/apache/comet/serde/maps.scala | 2 +-
.../main/scala/org/apache/comet/serde/math.scala | 15 +++-
.../scala/org/apache/comet/serde/predicates.scala | 2 +-
.../org/apache/comet/CometExpressionSuite.scala | 35 +++++++++
15 files changed, 148 insertions(+), 50 deletions(-)
diff --git a/docs/source/user-guide/latest/expressions.md
b/docs/source/user-guide/latest/expressions.md
index e727642dc..a255369ad 100644
--- a/docs/source/user-guide/latest/expressions.md
+++ b/docs/source/user-guide/latest/expressions.md
@@ -116,7 +116,7 @@ incompatible expressions.
## Math Expressions
| Expression | SQL | Spark-Compatible? | Compatibility Notes
|
-| -------------- | --------- | ----------------- |
--------------------------------- |
+| -------------- | --------- | -----------------
|-----------------------------------|
| Acos | `acos` | Yes |
|
| Add | `+` | Yes |
|
| Asin | `asin` | Yes |
|
@@ -140,7 +140,7 @@ incompatible expressions.
| Rand | `rand` | Yes |
|
| Randn | `randn` | Yes |
|
| Remainder | `%` | Yes | ANSI mode is not supported.
|
-| Round | `round` | Yes | ANSI mode is not supported.
|
+| Round | `round` | Yes |
|
| Signum | `signum` | Yes |
|
| Sin | `sin` | Yes |
|
| Sqrt | `sqrt` | Yes |
|
diff --git a/native/core/src/execution/planner.rs
b/native/core/src/execution/planner.rs
index 59090a5bd..1550efd79 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -2478,7 +2478,7 @@ impl PhysicalPlanner {
fun_name,
data_type.clone(),
&self.session_ctx.state(),
- None,
+ Some(expr.fail_on_error),
)?;
let args = args
@@ -3346,6 +3346,7 @@ mod tests {
func: "make_array".to_string(),
args: vec![array_col, array_col_1],
return_type: None,
+ fail_on_error: false,
})),
}],
})),
@@ -3464,6 +3465,7 @@ mod tests {
func: "array_repeat".to_string(),
args: vec![array_col, array_col_1],
return_type: None,
+ fail_on_error: false,
})),
}],
})),
diff --git a/native/proto/src/proto/expr.proto
b/native/proto/src/proto/expr.proto
index ade9860c8..5853bc613 100644
--- a/native/proto/src/proto/expr.proto
+++ b/native/proto/src/proto/expr.proto
@@ -308,6 +308,7 @@ message ScalarFunc {
string func = 1;
repeated Expr args = 2;
DataType return_type = 3;
+ bool fail_on_error = 4;
}
message CaseWhen {
diff --git a/native/spark-expr/src/comet_scalar_funcs.rs
b/native/spark-expr/src/comet_scalar_funcs.rs
index 19fa11e64..fc0c096b1 100644
--- a/native/spark-expr/src/comet_scalar_funcs.rs
+++ b/native/spark-expr/src/comet_scalar_funcs.rs
@@ -99,6 +99,7 @@ pub fn create_comet_physical_fun_with_eval_mode(
fail_on_error: Option<bool>,
eval_mode: EvalMode,
) -> Result<Arc<ScalarUDF>, DataFusionError> {
+ let fail_on_error = fail_on_error.unwrap_or(false);
match fun_name {
"ceil" => {
make_comet_scalar_udf!("ceil", spark_ceil, data_type)
@@ -119,7 +120,7 @@ pub fn create_comet_physical_fun_with_eval_mode(
make_comet_scalar_udf!("lpad", func, without data_type)
}
"round" => {
- make_comet_scalar_udf!("round", spark_round, data_type)
+ make_comet_scalar_udf!("round", spark_round, data_type,
fail_on_error)
}
"unscaled_value" => {
let func = Arc::new(spark_unscaled_value);
@@ -177,7 +178,6 @@ pub fn create_comet_physical_fun_with_eval_mode(
}
"spark_modulo" => {
let func = Arc::new(spark_modulo);
- let fail_on_error = fail_on_error.unwrap_or(false);
make_comet_scalar_udf!("spark_modulo", func, without data_type,
fail_on_error)
}
_ => registry.udf(fun_name).map_err(|e| {
diff --git a/native/spark-expr/src/math_funcs/round.rs
b/native/spark-expr/src/math_funcs/round.rs
index 9191a88ca..d2cbe4f96 100644
--- a/native/spark-expr/src/math_funcs/round.rs
+++ b/native/spark-expr/src/math_funcs/round.rs
@@ -15,49 +15,81 @@
// specific language governing permissions and limitations
// under the License.
+use crate::arithmetic_overflow_error;
use crate::math_funcs::utils::{get_precision_scale, make_decimal_array,
make_decimal_scalar};
use arrow::array::{Array, ArrowNativeTypeOp};
use arrow::array::{Int16Array, Int32Array, Int64Array, Int8Array};
use arrow::datatypes::DataType;
+use arrow::error::ArrowError;
use datafusion::common::{exec_err, internal_err, DataFusionError, ScalarValue};
use datafusion::{functions::math::round::round, physical_plan::ColumnarValue};
use std::{cmp::min, sync::Arc};
macro_rules! integer_round {
- ($X:expr, $DIV:expr, $HALF:expr) => {{
+ ($X:expr, $DIV:expr, $HALF:expr, $FAIL_ON_ERROR:expr) => {{
let rem = $X % $DIV;
if rem <= -$HALF {
- ($X - rem).sub_wrapping($DIV)
+ if $FAIL_ON_ERROR {
+ ($X - rem).sub_checked($DIV).map_err(|_| {
+
ArrowError::ComputeError(arithmetic_overflow_error("integer").to_string())
+ })
+ } else {
+ Ok(($X - rem).sub_wrapping($DIV))
+ }
} else if rem >= $HALF {
- ($X - rem).add_wrapping($DIV)
+ if $FAIL_ON_ERROR {
+ ($X - rem).add_checked($DIV).map_err(|_| {
+
ArrowError::ComputeError(arithmetic_overflow_error("integer").to_string())
+ })
+ } else {
+ Ok(($X - rem).add_wrapping($DIV))
+ }
} else {
- $X - rem
+ if $FAIL_ON_ERROR {
+ $X.sub_checked(rem).map_err(|_| {
+
ArrowError::ComputeError(arithmetic_overflow_error("integer").to_string())
+ })
+ } else {
+ Ok($X.sub_wrapping(rem))
+ }
}
}};
}
macro_rules! round_integer_array {
- ($ARRAY:expr, $POINT:expr, $TYPE:ty, $NATIVE:ty) => {{
+ ($ARRAY:expr, $POINT:expr, $TYPE:ty, $NATIVE:ty, $FAIL_ON_ERROR:expr) => {{
let array = $ARRAY.as_any().downcast_ref::<$TYPE>().unwrap();
let ten: $NATIVE = 10;
let result: $TYPE = if let Some(div) = ten.checked_pow((-(*$POINT)) as
u32) {
let half = div / 2;
- arrow::compute::kernels::arity::unary(array, |x| integer_round!(x,
div, half))
+ arrow::compute::kernels::arity::try_unary(array, |x| {
+ integer_round!(x, div, half, $FAIL_ON_ERROR)
+ })?
} else {
- arrow::compute::kernels::arity::unary(array, |_| 0)
+ arrow::compute::kernels::arity::try_unary(array, |_| Ok(0))?
};
Ok(ColumnarValue::Array(Arc::new(result)))
}};
}
macro_rules! round_integer_scalar {
- ($SCALAR:expr, $POINT:expr, $TYPE:expr, $NATIVE:ty) => {{
+ ($SCALAR:expr, $POINT:expr, $TYPE:expr, $NATIVE:ty, $FAIL_ON_ERROR:expr)
=> {{
let ten: $NATIVE = 10;
if let Some(div) = ten.checked_pow((-(*$POINT)) as u32) {
let half = div / 2;
- Ok(ColumnarValue::Scalar($TYPE(
- $SCALAR.map(|x| integer_round!(x, div, half)),
- )))
+ let scalar_opt = match $SCALAR {
+ Some(x) => match integer_round!(x, div, half, $FAIL_ON_ERROR) {
+ Ok(v) => Some(v),
+ Err(e) => {
+ return Err(DataFusionError::ArrowError(
+ Box::from(e),
+ Some(DataFusionError::get_back_trace()),
+ ))
+ }
+ },
+ None => None,
+ };
+ Ok(ColumnarValue::Scalar($TYPE(scalar_opt)))
} else {
Ok(ColumnarValue::Scalar($TYPE(Some(0))))
}
@@ -68,6 +100,7 @@ macro_rules! round_integer_scalar {
pub fn spark_round(
args: &[ColumnarValue],
data_type: &DataType,
+ fail_on_error: bool,
) -> Result<ColumnarValue, DataFusionError> {
let value = &args[0];
let point = &args[1];
@@ -76,10 +109,18 @@ pub fn spark_round(
};
match value {
ColumnarValue::Array(array) => match array.data_type() {
- DataType::Int64 if *point < 0 => round_integer_array!(array,
point, Int64Array, i64),
- DataType::Int32 if *point < 0 => round_integer_array!(array,
point, Int32Array, i32),
- DataType::Int16 if *point < 0 => round_integer_array!(array,
point, Int16Array, i16),
- DataType::Int8 if *point < 0 => round_integer_array!(array, point,
Int8Array, i8),
+ DataType::Int64 if *point < 0 => {
+ round_integer_array!(array, point, Int64Array, i64,
fail_on_error)
+ }
+ DataType::Int32 if *point < 0 => {
+ round_integer_array!(array, point, Int32Array, i32,
fail_on_error)
+ }
+ DataType::Int16 if *point < 0 => {
+ round_integer_array!(array, point, Int16Array, i16,
fail_on_error)
+ }
+ DataType::Int8 if *point < 0 => {
+ round_integer_array!(array, point, Int8Array, i8,
fail_on_error)
+ }
DataType::Decimal128(_, scale) if *scale >= 0 => {
let f = decimal_round_f(scale, point);
let (precision, scale) = get_precision_scale(data_type);
@@ -93,16 +134,16 @@ pub fn spark_round(
},
ColumnarValue::Scalar(a) => match a {
ScalarValue::Int64(a) if *point < 0 => {
- round_integer_scalar!(a, point, ScalarValue::Int64, i64)
+ round_integer_scalar!(a, point, ScalarValue::Int64, i64,
fail_on_error)
}
ScalarValue::Int32(a) if *point < 0 => {
- round_integer_scalar!(a, point, ScalarValue::Int32, i32)
+ round_integer_scalar!(a, point, ScalarValue::Int32, i32,
fail_on_error)
}
ScalarValue::Int16(a) if *point < 0 => {
- round_integer_scalar!(a, point, ScalarValue::Int16, i16)
+ round_integer_scalar!(a, point, ScalarValue::Int16, i16,
fail_on_error)
}
ScalarValue::Int8(a) if *point < 0 => {
- round_integer_scalar!(a, point, ScalarValue::Int8, i8)
+ round_integer_scalar!(a, point, ScalarValue::Int8, i8,
fail_on_error)
}
ScalarValue::Decimal128(a, _, scale) if *scale >= 0 => {
let f = decimal_round_f(scale, point);
@@ -158,7 +199,7 @@ mod test {
]))),
ColumnarValue::Scalar(ScalarValue::Int64(Some(2))),
];
- let ColumnarValue::Array(result) = spark_round(&args,
&DataType::Float32)? else {
+ let ColumnarValue::Array(result) = spark_round(&args,
&DataType::Float32, false)? else {
unreachable!()
};
let floats = as_float32_array(&result)?;
@@ -176,7 +217,7 @@ mod test {
]))),
ColumnarValue::Scalar(ScalarValue::Int64(Some(2))),
];
- let ColumnarValue::Array(result) = spark_round(&args,
&DataType::Float64)? else {
+ let ColumnarValue::Array(result) = spark_round(&args,
&DataType::Float64, false)? else {
unreachable!()
};
let floats = as_float64_array(&result)?;
@@ -193,7 +234,7 @@ mod test {
ColumnarValue::Scalar(ScalarValue::Int64(Some(2))),
];
let ColumnarValue::Scalar(ScalarValue::Float32(Some(result))) =
- spark_round(&args, &DataType::Float32)?
+ spark_round(&args, &DataType::Float32, false)?
else {
unreachable!()
};
@@ -209,7 +250,7 @@ mod test {
ColumnarValue::Scalar(ScalarValue::Int64(Some(2))),
];
let ColumnarValue::Scalar(ScalarValue::Float64(Some(result))) =
- spark_round(&args, &DataType::Float64)?
+ spark_round(&args, &DataType::Float64, false)?
else {
unreachable!()
};
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index 0e91cd41f..afde79599 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -851,7 +851,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
case UnscaledValue(child) =>
val childExpr = exprToProtoInternal(child, inputs, binding)
val optExpr =
- scalarFunctionExprToProtoWithReturnType("unscaled_value", LongType,
childExpr)
+ scalarFunctionExprToProtoWithReturnType("unscaled_value", LongType,
false, childExpr)
optExprWithInfo(optExpr, expr, child)
case MakeDecimal(child, precision, scale, true) =>
@@ -859,6 +859,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
val optExpr = scalarFunctionExprToProtoWithReturnType(
"make_decimal",
DecimalType(precision, scale),
+ false,
childExpr)
optExprWithInfo(optExpr, expr, child)
@@ -967,9 +968,11 @@ object QueryPlanSerde extends Logging with CometExprShim {
def scalarFunctionExprToProtoWithReturnType(
funcName: String,
returnType: DataType,
+ failOnError: Boolean,
args: Option[Expr]*): Option[Expr] = {
val builder = ExprOuterClass.ScalarFunc.newBuilder()
builder.setFunc(funcName)
+ builder.setFailOnError(failOnError)
serializeDataType(returnType).flatMap { t =>
builder.setReturnType(t)
scalarFunctionExprToProto0(builder, args: _*)
@@ -979,6 +982,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
def scalarFunctionExprToProto(funcName: String, args: Option[Expr]*):
Option[Expr] = {
val builder = ExprOuterClass.ScalarFunc.newBuilder()
builder.setFunc(funcName)
+ builder.setFailOnError(false)
scalarFunctionExprToProto0(builder, args: _*)
}
diff --git a/spark/src/main/scala/org/apache/comet/serde/arithmetic.scala
b/spark/src/main/scala/org/apache/comet/serde/arithmetic.scala
index eed4d7b16..a97a87e73 100644
--- a/spark/src/main/scala/org/apache/comet/serde/arithmetic.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/arithmetic.scala
@@ -279,14 +279,6 @@ object CometRemainder extends
CometExpressionSerde[Remainder] with MathBase {
object CometRound extends CometExpressionSerde[Round] {
- override def getSupportLevel(expr: Round): SupportLevel = {
- if (expr.ansiEnabled) {
- Incompatible(Some("ANSI mode is not supported"))
- } else {
- Compatible(None)
- }
- }
-
override def convert(
r: Round,
inputs: Seq[Attribute],
@@ -325,7 +317,12 @@ object CometRound extends CometExpressionSerde[Round] {
// `scale` must be Int64 type in DataFusion
val scaleExpr = exprToProtoInternal(Literal(_scale.toLong, LongType),
inputs, binding)
val optExpr =
- scalarFunctionExprToProtoWithReturnType("round", r.dataType,
childExpr, scaleExpr)
+ scalarFunctionExprToProtoWithReturnType(
+ "round",
+ r.dataType,
+ r.ansiEnabled,
+ childExpr,
+ scaleExpr)
optExprWithInfo(optExpr, r, r.child)
}
diff --git a/spark/src/main/scala/org/apache/comet/serde/arrays.scala
b/spark/src/main/scala/org/apache/comet/serde/arrays.scala
index 334f8ee8b..e8a4ce76f 100644
--- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala
@@ -210,6 +210,7 @@ object CometArraysOverlap extends
CometExpressionSerde[ArraysOverlap] {
val arraysOverlapScalarExpr = scalarFunctionExprToProtoWithReturnType(
"array_has_any",
BooleanType,
+ false,
leftArrayExprProto,
rightArrayExprProto)
optExprWithInfo(arraysOverlapScalarExpr, expr, expr.children: _*)
@@ -250,6 +251,7 @@ object CometArrayCompact extends
CometExpressionSerde[Expression] {
val arrayCompactScalarExpr = scalarFunctionExprToProtoWithReturnType(
"array_remove_all",
ArrayType(elementType = elementType),
+ false,
arrayExprProto,
nullLiteralProto)
optExprWithInfo(arrayCompactScalarExpr, expr, expr.children: _*)
diff --git a/spark/src/main/scala/org/apache/comet/serde/bitwise.scala
b/spark/src/main/scala/org/apache/comet/serde/bitwise.scala
index 8215ea5df..56404fe3c 100644
--- a/spark/src/main/scala/org/apache/comet/serde/bitwise.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/bitwise.scala
@@ -135,7 +135,7 @@ object CometBitwiseGet extends
CometExpressionSerde[BitwiseGet] {
val argProto = exprToProto(expr.left, inputs, binding)
val posProto = exprToProto(expr.right, inputs, binding)
val bitGetScalarExpr =
- scalarFunctionExprToProtoWithReturnType("bit_get", ByteType, argProto,
posProto)
+ scalarFunctionExprToProtoWithReturnType("bit_get", ByteType, false,
argProto, posProto)
optExprWithInfo(bitGetScalarExpr, expr, expr.children: _*)
}
}
@@ -147,7 +147,7 @@ object CometBitwiseCount extends
CometExpressionSerde[BitwiseCount] {
binding: Boolean): Option[ExprOuterClass.Expr] = {
val childProto = exprToProto(expr.child, inputs, binding)
val bitCountScalarExpr =
- scalarFunctionExprToProtoWithReturnType("bit_count", IntegerType,
childProto)
+ scalarFunctionExprToProtoWithReturnType("bit_count", IntegerType, false,
childProto)
optExprWithInfo(bitCountScalarExpr, expr, expr.children: _*)
}
}
diff --git a/spark/src/main/scala/org/apache/comet/serde/datetime.scala
b/spark/src/main/scala/org/apache/comet/serde/datetime.scala
index 9473ee30e..267b2b441 100644
--- a/spark/src/main/scala/org/apache/comet/serde/datetime.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/datetime.scala
@@ -263,7 +263,12 @@ object CometTruncDate extends
CometExpressionSerde[TruncDate] {
val childExpr = exprToProtoInternal(expr.date, inputs, binding)
val formatExpr = exprToProtoInternal(expr.format, inputs, binding)
val optExpr =
- scalarFunctionExprToProtoWithReturnType("date_trunc", DateType,
childExpr, formatExpr)
+ scalarFunctionExprToProtoWithReturnType(
+ "date_trunc",
+ DateType,
+ false,
+ childExpr,
+ formatExpr)
optExprWithInfo(optExpr, expr, expr.date, expr.format)
}
}
diff --git a/spark/src/main/scala/org/apache/comet/serde/hash.scala
b/spark/src/main/scala/org/apache/comet/serde/hash.scala
index 5c45a2593..a384ecf44 100644
--- a/spark/src/main/scala/org/apache/comet/serde/hash.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/hash.scala
@@ -40,7 +40,7 @@ object CometXxHash64 extends CometExpressionSerde[XxHash64] {
.setLongVal(expr.seed)
val seedExpr =
Some(ExprOuterClass.Expr.newBuilder().setLiteral(seedBuilder).build())
// the seed is put at the end of the arguments
- scalarFunctionExprToProtoWithReturnType("xxhash64", LongType, exprs :+
seedExpr: _*)
+ scalarFunctionExprToProtoWithReturnType("xxhash64", LongType, false, exprs
:+ seedExpr: _*)
}
}
@@ -59,7 +59,11 @@ object CometMurmur3Hash extends
CometExpressionSerde[Murmur3Hash] {
.setIntVal(expr.seed)
val seedExpr =
Some(ExprOuterClass.Expr.newBuilder().setLiteral(seedBuilder).build())
// the seed is put at the end of the arguments
- scalarFunctionExprToProtoWithReturnType("murmur3_hash", IntegerType, exprs
:+ seedExpr: _*)
+ scalarFunctionExprToProtoWithReturnType(
+ "murmur3_hash",
+ IntegerType,
+ false,
+ exprs :+ seedExpr: _*)
}
}
@@ -81,7 +85,7 @@ object CometSha2 extends CometExpressionSerde[Sha2] {
val leftExpr = exprToProtoInternal(expr.left, inputs, binding)
val numBitsExpr = exprToProtoInternal(expr.right, inputs, binding)
- scalarFunctionExprToProtoWithReturnType("sha2", StringType, leftExpr,
numBitsExpr)
+ scalarFunctionExprToProtoWithReturnType("sha2", StringType, false,
leftExpr, numBitsExpr)
}
}
diff --git a/spark/src/main/scala/org/apache/comet/serde/maps.scala
b/spark/src/main/scala/org/apache/comet/serde/maps.scala
index 572ac37d4..2e217f6af 100644
--- a/spark/src/main/scala/org/apache/comet/serde/maps.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/maps.scala
@@ -85,7 +85,7 @@ object CometMapFromArrays extends
CometExpressionSerde[MapFromArrays] {
val valueType = expr.right.dataType.asInstanceOf[ArrayType].elementType
val returnType = MapType(keyType = keyType, valueType = valueType)
val mapFromArraysExpr =
- scalarFunctionExprToProtoWithReturnType("map", returnType, keysExpr,
valuesExpr)
+ scalarFunctionExprToProtoWithReturnType("map", returnType, false,
keysExpr, valuesExpr)
optExprWithInfo(mapFromArraysExpr, expr, expr.children: _*)
}
}
diff --git a/spark/src/main/scala/org/apache/comet/serde/math.scala
b/spark/src/main/scala/org/apache/comet/serde/math.scala
index f2b0010d8..bfcd242d7 100644
--- a/spark/src/main/scala/org/apache/comet/serde/math.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/math.scala
@@ -50,7 +50,8 @@ object CometCeil extends CometExpressionSerde[Ceil] {
withInfo(expr, s"Decimal type $t has negative scale")
None
case _ =>
- val optExpr = scalarFunctionExprToProtoWithReturnType("ceil",
expr.dataType, childExpr)
+ val optExpr =
+ scalarFunctionExprToProtoWithReturnType("ceil", expr.dataType,
false, childExpr)
optExprWithInfo(optExpr, expr, expr.child)
}
}
@@ -69,7 +70,8 @@ object CometFloor extends CometExpressionSerde[Floor] {
withInfo(expr, s"Decimal type $t has negative scale")
None
case _ =>
- val optExpr = scalarFunctionExprToProtoWithReturnType("floor",
expr.dataType, childExpr)
+ val optExpr =
+ scalarFunctionExprToProtoWithReturnType("floor", expr.dataType,
false, childExpr)
optExprWithInfo(optExpr, expr, expr.child)
}
}
@@ -118,7 +120,7 @@ object CometHex extends CometExpressionSerde[Hex] with
MathExprBase {
inputs: Seq[Attribute],
binding: Boolean): Option[ExprOuterClass.Expr] = {
val childExpr = exprToProtoInternal(expr.child, inputs, binding)
- val optExpr = scalarFunctionExprToProtoWithReturnType("hex",
expr.dataType, childExpr)
+ val optExpr = scalarFunctionExprToProtoWithReturnType("hex",
expr.dataType, false, childExpr)
optExprWithInfo(optExpr, expr, expr.child)
}
}
@@ -132,7 +134,12 @@ object CometUnhex extends CometExpressionSerde[Unhex] with
MathExprBase {
val failOnErrorExpr = exprToProtoInternal(Literal(expr.failOnError),
inputs, binding)
val optExpr =
- scalarFunctionExprToProtoWithReturnType("unhex", expr.dataType,
childExpr, failOnErrorExpr)
+ scalarFunctionExprToProtoWithReturnType(
+ "unhex",
+ expr.dataType,
+ false,
+ childExpr,
+ failOnErrorExpr)
optExprWithInfo(optExpr, expr, expr.child)
}
}
diff --git a/spark/src/main/scala/org/apache/comet/serde/predicates.scala
b/spark/src/main/scala/org/apache/comet/serde/predicates.scala
index 00a1fb6fa..2e253bc23 100644
--- a/spark/src/main/scala/org/apache/comet/serde/predicates.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/predicates.scala
@@ -218,7 +218,7 @@ object CometIsNaN extends CometExpressionSerde[IsNaN] {
inputs: Seq[Attribute],
binding: Boolean): Option[ExprOuterClass.Expr] = {
val childExpr = exprToProtoInternal(expr.child, inputs, binding)
- val optExpr = scalarFunctionExprToProtoWithReturnType("isnan",
BooleanType, childExpr)
+ val optExpr = scalarFunctionExprToProtoWithReturnType("isnan",
BooleanType, false, childExpr)
optExprWithInfo(optExpr, expr, expr.child)
}
diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
index 27c9124f1..9085c0fa2 100644
--- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
@@ -3017,6 +3017,41 @@ class CometExpressionSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
}
}
+ test("ANSI support for round function") {
+ Seq(
+ (
+ Int.MaxValue,
+ Int.MinValue,
+ Long.MinValue,
+ Long.MaxValue,
+ Byte.MinValue,
+ Byte.MaxValue,
+ Short.MinValue,
+ Short.MaxValue)).foreach { value =>
+ val data = Seq(value)
+ withParquetTable(data, "tbl") {
+ Seq(-1000, -100, -10, -1, 0, 1, 10, 100, 1000).foreach { scale =>
+ Seq(true, false).foreach { ansi =>
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansi.toString) {
+ val res = spark.sql(s"SELECT round(_1, $scale) from tbl")
+ checkSparkMaybeThrows(res) match {
+ case (Some(sparkException), Some(cometException)) =>
+
assert(sparkException.getMessage.contains("ARITHMETIC_OVERFLOW"))
+
assert(cometException.getMessage.contains("ARITHMETIC_OVERFLOW"))
+ case (None, None) => checkSparkAnswerAndOperator(res)
+ case (None, Some(ex)) =>
+ fail("Comet threw an exception but Spark did not. Comet
exception: " + ex.getMessage)
+ case (Some(sparkException), None) =>
+ fail("Spark threw an exception but Comet did not. Spark
exception: " +
+ sparkException.getMessage)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
test("test integral divide overflow for decimal") {
if (isSpark40Plus) {
Seq(true, false)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
