You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "irenjj (via GitHub)" <gi...@apache.org> on 2025/05/06 08:35:15 UTC
[PR] Implement ceil&floor function for spark [datafusion]
irenjj opened a new pull request, #15958:
URL: https://github.com/apache/datafusion/pull/15958
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
- Closes #15916
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
## Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement ceil&floor function for spark [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#issuecomment-2860681100
@parthchandra @huaxingao @comphead @mbutrovich @kazuyukitanimura, FYI, in case you want to review. In Comet, we are currently using DataFusion's ceil and floor functions and have not specialized for Spark.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#discussion_r2078773309
##########
datafusion/spark/src/function/math/ceil_floor.rs:
##########
@@ -0,0 +1,720 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use arrow::{
+ array::{ArrayRef, ArrowNativeTypeOp, AsArray},
+ datatypes::{
+ DataType, Decimal128Type, Field, Float32Type, Float64Type, Int16Type, Int32Type,
+ Int64Type, Int8Type, DECIMAL128_MAX_PRECISION, DECIMAL128_MAX_SCALE,
+ },
+};
+use datafusion_common::{Result, ScalarValue};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+use num::integer::{div_ceil, div_floor};
+
+use crate::function::error_utils::{
+ generic_exec_err, generic_internal_err, invalid_arg_count_exec_err,
+ unsupported_data_type_exec_err, unsupported_data_types_exec_err,
+};
+
+use super::round_decimal_base;
+
+fn ceil_floor_coerce_types(name: &str, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ if arg_types.len() == 1 {
+ if arg_types[0].is_numeric() {
+ Ok(vec![ceil_floor_coerce_first_arg(name, &arg_types[0])?])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type",
+ arg_types,
+ ))
+ }
+ } else if arg_types.len() == 2 {
+ if arg_types[0].is_numeric() && arg_types[1].is_integer() {
+ Ok(vec![
+ ceil_floor_coerce_first_arg(name, &arg_types[0])?,
+ DataType::Int32,
+ ])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type for expr and Integer Type for target scale",
+ arg_types,
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_types.len()))
+ }
+}
+
+fn ceil_floor_return_field_from_args(name: &str, args: ReturnFieldArgs) -> Result<Field> {
+ let arg_fields = args.arg_fields;
+ let scalar_arguments = args.scalar_arguments;
+ let return_type = if arg_fields.len() == 1 {
+ match &arg_fields[0].data_type() {
+ DataType::Decimal128(precision, scale) => {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, true);
+ Ok(DataType::Decimal128(precision, scale))
+ }
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, false);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Ok(DataType::Int64),
+ }
+ } else if arg_fields.len() == 2 {
+ if let Some(target_scale) = scalar_arguments[1] {
+ let expr = &arg_fields[0].data_type();
+ let target_scale: i32 = match target_scale {
+ ScalarValue::Int8(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int16(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int32(Some(v)) => Ok(*v),
+ ScalarValue::Int64(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt8(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt16(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt32(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt64(Some(v)) => Ok(*v as i32),
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Target scale must be Integer literal",
+ &target_scale.data_type(),
+ )),
+ }?;
+ if target_scale < -38 {
+ return Err(generic_exec_err(
+ name,
+ "Target scale must be greater than -38",
+ ));
+ }
+ let (precision, scale) = match expr {
+ DataType::Int8 => Ok((3, 0)),
+ DataType::UInt8 | DataType::Int16 => Ok((5, 0)),
+ DataType::UInt16 | DataType::Int32 => Ok((10, 0)),
+ DataType::UInt32 | DataType::UInt64 | DataType::Int64 => Ok((20, 0)),
+ DataType::Float32 => Ok((14, 7)),
+ DataType::Float64 => Ok((30, 15)),
+ DataType::Decimal128(precision, scale)
+ | DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok((*precision as i32, *scale as i32))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Numeric Type for expr",
+ expr,
+ )),
+ }?;
+ let (precision, scale) =
+ round_decimal_base(precision, scale, target_scale, true);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(generic_exec_err(
+ name,
+ "Target scale must be Integer literal, received: None",
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_fields.len()))
+ }?;
+ Ok(Field::new(name.to_string(), return_type, true))
+}
+
+fn ceil_floor_coerce_first_arg(name: &str, arg_type: &DataType) -> Result<DataType> {
+ if arg_type.is_numeric() {
+ match arg_type {
+ DataType::UInt8 => Ok(DataType::Int16),
+ DataType::UInt16 => Ok(DataType::Int32),
+ DataType::UInt32 | DataType::UInt64 => Ok(DataType::Int64),
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok(DataType::Decimal128(*precision, *scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_type,
+ ))
+ }
+ }
+ other => Ok(other.clone()),
+ }
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ "First arg must be Numeric Type",
+ arg_type,
+ ))
+ }
+}
+
+#[inline]
+fn get_return_type_precision_scale(return_type: &DataType) -> Result<(u8, i8)> {
+ match return_type {
+ DataType::Decimal128(precision, scale) => Ok((*precision, *scale)),
+ other => Err(generic_internal_err(
+ "ceil",
+ format!("Expected return type to be Decimal128, got: {other}").as_str(),
+ )),
+ }
+}
+
+#[derive(Debug)]
+pub struct SparkCeil {
+ signature: Signature,
+}
+
+impl Default for SparkCeil {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkCeil {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkCeil {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn name(&self) -> &str {
+ "ceil"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Err(generic_internal_err(
+ "ceil",
+ "`return_type` should not be called, call `return_type_from_args` instead",
+ ))
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<Field> {
+ ceil_floor_return_field_from_args("ceil", args)
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ let arg_len = args.args.len();
+ let target_scale = if arg_len == 1 {
+ Ok(&None)
+ } else if arg_len == 2 {
Review Comment:
Ah, there is a separate `RoundCeil` in Spark for the two-argument case:
```scala
case class RoundCeil(child: Expression, scale: Expression)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "irenjj (via GitHub)" <gi...@apache.org>.
irenjj commented on code in PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#discussion_r2078862746
##########
datafusion/spark/src/function/math/ceil_floor.rs:
##########
@@ -0,0 +1,720 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use arrow::{
+ array::{ArrayRef, ArrowNativeTypeOp, AsArray},
+ datatypes::{
+ DataType, Decimal128Type, Field, Float32Type, Float64Type, Int16Type, Int32Type,
+ Int64Type, Int8Type, DECIMAL128_MAX_PRECISION, DECIMAL128_MAX_SCALE,
+ },
+};
+use datafusion_common::{Result, ScalarValue};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+use num::integer::{div_ceil, div_floor};
+
+use crate::function::error_utils::{
+ generic_exec_err, generic_internal_err, invalid_arg_count_exec_err,
+ unsupported_data_type_exec_err, unsupported_data_types_exec_err,
+};
+
+use super::round_decimal_base;
+
+fn ceil_floor_coerce_types(name: &str, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ if arg_types.len() == 1 {
+ if arg_types[0].is_numeric() {
+ Ok(vec![ceil_floor_coerce_first_arg(name, &arg_types[0])?])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type",
+ arg_types,
+ ))
+ }
+ } else if arg_types.len() == 2 {
+ if arg_types[0].is_numeric() && arg_types[1].is_integer() {
+ Ok(vec![
+ ceil_floor_coerce_first_arg(name, &arg_types[0])?,
+ DataType::Int32,
+ ])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type for expr and Integer Type for target scale",
+ arg_types,
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_types.len()))
+ }
+}
+
+fn ceil_floor_return_field_from_args(name: &str, args: ReturnFieldArgs) -> Result<Field> {
+ let arg_fields = args.arg_fields;
+ let scalar_arguments = args.scalar_arguments;
+ let return_type = if arg_fields.len() == 1 {
+ match &arg_fields[0].data_type() {
+ DataType::Decimal128(precision, scale) => {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, true);
+ Ok(DataType::Decimal128(precision, scale))
+ }
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, false);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Ok(DataType::Int64),
+ }
+ } else if arg_fields.len() == 2 {
+ if let Some(target_scale) = scalar_arguments[1] {
+ let expr = &arg_fields[0].data_type();
+ let target_scale: i32 = match target_scale {
+ ScalarValue::Int8(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int16(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int32(Some(v)) => Ok(*v),
+ ScalarValue::Int64(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt8(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt16(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt32(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt64(Some(v)) => Ok(*v as i32),
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Target scale must be Integer literal",
+ &target_scale.data_type(),
+ )),
+ }?;
+ if target_scale < -38 {
+ return Err(generic_exec_err(
+ name,
+ "Target scale must be greater than -38",
+ ));
+ }
+ let (precision, scale) = match expr {
+ DataType::Int8 => Ok((3, 0)),
+ DataType::UInt8 | DataType::Int16 => Ok((5, 0)),
+ DataType::UInt16 | DataType::Int32 => Ok((10, 0)),
+ DataType::UInt32 | DataType::UInt64 | DataType::Int64 => Ok((20, 0)),
+ DataType::Float32 => Ok((14, 7)),
+ DataType::Float64 => Ok((30, 15)),
+ DataType::Decimal128(precision, scale)
+ | DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok((*precision as i32, *scale as i32))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Numeric Type for expr",
+ expr,
+ )),
+ }?;
+ let (precision, scale) =
+ round_decimal_base(precision, scale, target_scale, true);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(generic_exec_err(
+ name,
+ "Target scale must be Integer literal, received: None",
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_fields.len()))
+ }?;
+ Ok(Field::new(name.to_string(), return_type, true))
+}
+
+fn ceil_floor_coerce_first_arg(name: &str, arg_type: &DataType) -> Result<DataType> {
+ if arg_type.is_numeric() {
+ match arg_type {
+ DataType::UInt8 => Ok(DataType::Int16),
+ DataType::UInt16 => Ok(DataType::Int32),
+ DataType::UInt32 | DataType::UInt64 => Ok(DataType::Int64),
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok(DataType::Decimal128(*precision, *scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_type,
+ ))
+ }
+ }
+ other => Ok(other.clone()),
+ }
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ "First arg must be Numeric Type",
+ arg_type,
+ ))
+ }
+}
+
+#[inline]
+fn get_return_type_precision_scale(return_type: &DataType) -> Result<(u8, i8)> {
+ match return_type {
+ DataType::Decimal128(precision, scale) => Ok((*precision, *scale)),
+ other => Err(generic_internal_err(
+ "ceil",
+ format!("Expected return type to be Decimal128, got: {other}").as_str(),
+ )),
+ }
+}
+
+#[derive(Debug)]
+pub struct SparkCeil {
+ signature: Signature,
+}
+
+impl Default for SparkCeil {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkCeil {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkCeil {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn name(&self) -> &str {
+ "ceil"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Err(generic_internal_err(
+ "ceil",
+ "`return_type` should not be called, call `return_type_from_args` instead",
+ ))
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<Field> {
+ ceil_floor_return_field_from_args("ceil", args)
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ let arg_len = args.args.len();
+ let target_scale = if arg_len == 1 {
+ Ok(&None)
+ } else if arg_len == 2 {
Review Comment:
Got it. In this PR, we only need to implement the single-argument version of `ceil`, and add the new function `roundceil` as a two-argument implementation in a future PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "shehabgamin (via GitHub)" <gi...@apache.org>.
shehabgamin commented on code in PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#discussion_r2080822728
##########
datafusion/sqllogictest/test_files/spark/math/ceil.slt:
##########
@@ -0,0 +1,141 @@
+# 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.
+
+query R
+SELECT ceil(-0.1);
+----
+0
+
+query R
+SELECT ceil(5.4);
+----
+6
+
+query error
+SELECT ceil(3.1411, -3);
+----
+DataFusion error: Error during planning: Failed to coerce arguments to satisfy a call to 'ceil' function: coercion from [Float64, Int64] to the signature Uniform(1, [Float64, Float32]) failed
Review Comment:
@irenjj Sail doesn't use the same SQL parser as DataFusion, as it follows Spark SQL syntax.
You’ll need to cast values to ensure the tests interpret the correct `DataType`.
For example, instead of this:
```
SELECT ceil(3.1411, -3);
```
Use:
```
SELECT ceil(3.1411::decimal(5,4), -3::INT);
```
Copy-pasting the exact queries from the files below won't work directly. However, if you look closely, you’ll see that each query includes the necessary cast for the first argument (the second argument is always an `INT`):
- https://github.com/lakehq/sail/blob/main/python/pysail/tests/spark/function/test_ceil.txt
- https://github.com/lakehq/sail/blob/main/python/pysail/tests/spark/function/test_floor.txt
However, you should expect the same result values from those two files.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#discussion_r2078769590
##########
datafusion/spark/src/function/math/ceil_floor.rs:
##########
@@ -0,0 +1,720 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use arrow::{
+ array::{ArrayRef, ArrowNativeTypeOp, AsArray},
+ datatypes::{
+ DataType, Decimal128Type, Field, Float32Type, Float64Type, Int16Type, Int32Type,
+ Int64Type, Int8Type, DECIMAL128_MAX_PRECISION, DECIMAL128_MAX_SCALE,
+ },
+};
+use datafusion_common::{Result, ScalarValue};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+use num::integer::{div_ceil, div_floor};
+
+use crate::function::error_utils::{
+ generic_exec_err, generic_internal_err, invalid_arg_count_exec_err,
+ unsupported_data_type_exec_err, unsupported_data_types_exec_err,
+};
+
+use super::round_decimal_base;
+
+fn ceil_floor_coerce_types(name: &str, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ if arg_types.len() == 1 {
+ if arg_types[0].is_numeric() {
+ Ok(vec![ceil_floor_coerce_first_arg(name, &arg_types[0])?])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type",
+ arg_types,
+ ))
+ }
+ } else if arg_types.len() == 2 {
+ if arg_types[0].is_numeric() && arg_types[1].is_integer() {
+ Ok(vec![
+ ceil_floor_coerce_first_arg(name, &arg_types[0])?,
+ DataType::Int32,
+ ])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type for expr and Integer Type for target scale",
+ arg_types,
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_types.len()))
+ }
+}
+
+fn ceil_floor_return_field_from_args(name: &str, args: ReturnFieldArgs) -> Result<Field> {
+ let arg_fields = args.arg_fields;
+ let scalar_arguments = args.scalar_arguments;
+ let return_type = if arg_fields.len() == 1 {
+ match &arg_fields[0].data_type() {
+ DataType::Decimal128(precision, scale) => {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, true);
+ Ok(DataType::Decimal128(precision, scale))
+ }
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, false);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Ok(DataType::Int64),
+ }
+ } else if arg_fields.len() == 2 {
+ if let Some(target_scale) = scalar_arguments[1] {
+ let expr = &arg_fields[0].data_type();
+ let target_scale: i32 = match target_scale {
+ ScalarValue::Int8(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int16(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int32(Some(v)) => Ok(*v),
+ ScalarValue::Int64(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt8(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt16(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt32(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt64(Some(v)) => Ok(*v as i32),
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Target scale must be Integer literal",
+ &target_scale.data_type(),
+ )),
+ }?;
+ if target_scale < -38 {
+ return Err(generic_exec_err(
+ name,
+ "Target scale must be greater than -38",
+ ));
+ }
+ let (precision, scale) = match expr {
+ DataType::Int8 => Ok((3, 0)),
+ DataType::UInt8 | DataType::Int16 => Ok((5, 0)),
+ DataType::UInt16 | DataType::Int32 => Ok((10, 0)),
+ DataType::UInt32 | DataType::UInt64 | DataType::Int64 => Ok((20, 0)),
+ DataType::Float32 => Ok((14, 7)),
+ DataType::Float64 => Ok((30, 15)),
+ DataType::Decimal128(precision, scale)
+ | DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok((*precision as i32, *scale as i32))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Numeric Type for expr",
+ expr,
+ )),
+ }?;
+ let (precision, scale) =
+ round_decimal_base(precision, scale, target_scale, true);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(generic_exec_err(
+ name,
+ "Target scale must be Integer literal, received: None",
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_fields.len()))
+ }?;
+ Ok(Field::new(name.to_string(), return_type, true))
+}
+
+fn ceil_floor_coerce_first_arg(name: &str, arg_type: &DataType) -> Result<DataType> {
+ if arg_type.is_numeric() {
+ match arg_type {
+ DataType::UInt8 => Ok(DataType::Int16),
+ DataType::UInt16 => Ok(DataType::Int32),
+ DataType::UInt32 | DataType::UInt64 => Ok(DataType::Int64),
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok(DataType::Decimal128(*precision, *scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_type,
+ ))
+ }
+ }
+ other => Ok(other.clone()),
+ }
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ "First arg must be Numeric Type",
+ arg_type,
+ ))
+ }
+}
+
+#[inline]
+fn get_return_type_precision_scale(return_type: &DataType) -> Result<(u8, i8)> {
+ match return_type {
+ DataType::Decimal128(precision, scale) => Ok((*precision, *scale)),
+ other => Err(generic_internal_err(
+ "ceil",
+ format!("Expected return type to be Decimal128, got: {other}").as_str(),
+ )),
+ }
+}
+
+#[derive(Debug)]
+pub struct SparkCeil {
+ signature: Signature,
+}
+
+impl Default for SparkCeil {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkCeil {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkCeil {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn name(&self) -> &str {
+ "ceil"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Err(generic_internal_err(
+ "ceil",
+ "`return_type` should not be called, call `return_type_from_args` instead",
+ ))
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<Field> {
+ ceil_floor_return_field_from_args("ceil", args)
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ let arg_len = args.args.len();
+ let target_scale = if arg_len == 1 {
+ Ok(&None)
+ } else if arg_len == 2 {
Review Comment:
Floor and Ceil expressions in Spark are unary expressions and only take a single argument:
```scala
case class Ceil(child: _root_.org.apache.spark.sql.catalyst.expressions.Expression) extends _root_.org.apache.spark.sql.catalyst.expressions.UnaryMathExpression {
```
When calling `ceil` in Spark SQL with two parameters, it looks like a cast is added to the input expression. For example, `ceil(_1, 2)` translates to `plan: *(1) Project [ceil(cast(_1#6 as decimal(30,15)), 2) AS ceil(_1, 2)#10`.
Perhaps you implemented the two-argument version to work around the fact that DataFusion's logical optimizer doesn't implement the same logic as Spark for adding this cast? I wonder if it is worth looking into adding a Spark-compatible optimizer rule to add this cast in the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement ceil&floor function for spark [datafusion]
Posted by "shehabgamin (via GitHub)" <gi...@apache.org>.
shehabgamin commented on PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#issuecomment-2856156155
@irenjj Here are some tests for `ceil` and `floor` that you can re-use. Sail adheres to Spark SQL, so you may need to adjust as needed for this pr:
- https://github.com/lakehq/sail/blob/main/python/pysail/tests/spark/function/test_ceil.txt
- https://github.com/lakehq/sail/blob/main/python/pysail/tests/spark/function/test_floor.txt
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#issuecomment-2982055315
@irenjj - I wonder if you would be willing to help pick this PR back up now that we have merged a PR with a bunch of tests from @shehabgamin here:
- https://github.com/apache/datafusion/pull/16409
We are hoping to generate some example PRs to use when starting to fill out the rest of the functions and I was thinking this one might be a good one
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement ceil&floor function for spark [datafusion]
Posted by "shehabgamin (via GitHub)" <gi...@apache.org>.
shehabgamin commented on code in PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#discussion_r2076361890
##########
datafusion/spark/src/function/math/ceil_floor.rs:
##########
@@ -0,0 +1,716 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use arrow::{
+ array::{ArrayRef, ArrowNativeTypeOp, AsArray},
+ datatypes::{
+ DataType, Decimal128Type, Float32Type, Float64Type, Int16Type, Int32Type,
+ Int64Type, Int8Type, DECIMAL128_MAX_PRECISION, DECIMAL128_MAX_SCALE,
+ },
+};
+use datafusion_common::{Result, ScalarValue};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+use num::integer::{div_ceil, div_floor};
+
+use crate::function::error_utils::{
+ generic_exec_err, generic_internal_err, invalid_arg_count_exec_err,
+ unsupported_data_type_exec_err, unsupported_data_types_exec_err,
+};
+
+use super::round_decimal_base;
+
+fn ceil_floor_coerce_types(name: &str, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ if arg_types.len() == 1 {
+ if arg_types[0].is_numeric() {
+ Ok(vec![ceil_floor_coerce_first_arg(name, &arg_types[0])?])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type",
+ arg_types,
+ ))
+ }
+ } else if arg_types.len() == 2 {
+ if arg_types[0].is_numeric() && arg_types[1].is_integer() {
+ Ok(vec![
+ ceil_floor_coerce_first_arg(name, &arg_types[0])?,
+ DataType::Int32,
+ ])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type for expr and Integer Type for target scale",
+ arg_types,
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_types.len()))
+ }
+}
+
+#[allow(dead_code)]
+fn ceil_floor_return_type_from_args(
+ name: &str,
+ args: ReturnFieldArgs,
+) -> Result<DataType> {
+ let arg_fields = args.arg_fields;
+ let scalar_arguments = args.scalar_arguments;
+ let return_type = if arg_fields.len() == 1 {
+ match &arg_fields[0].data_type() {
+ DataType::Decimal128(precision, scale) => {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, true);
+ Ok(DataType::Decimal128(precision, scale))
+ }
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, false);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ &arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Ok(DataType::Int64),
+ }
+ } else if arg_fields.len() == 2 {
+ if let Some(target_scale) = scalar_arguments[1] {
+ let expr = &arg_fields[0].data_type();
+ let target_scale: i32 = match target_scale {
+ ScalarValue::Int8(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int16(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int32(Some(v)) => Ok(*v),
+ ScalarValue::Int64(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt8(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt16(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt32(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt64(Some(v)) => Ok(*v as i32),
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Target scale must be Integer literal",
+ &target_scale.data_type(),
+ )),
+ }?;
+ if target_scale < -38 {
+ return Err(generic_exec_err(
+ name,
+ "Target scale must be greater than -38",
+ ));
+ }
+ let (precision, scale) = match expr {
+ DataType::Int8 => Ok((3, 0)),
+ DataType::UInt8 | DataType::Int16 => Ok((5, 0)),
+ DataType::UInt16 | DataType::Int32 => Ok((10, 0)),
+ DataType::UInt32 | DataType::UInt64 | DataType::Int64 => Ok((20, 0)),
+ DataType::Float32 => Ok((14, 7)),
+ DataType::Float64 => Ok((30, 15)),
+ DataType::Decimal128(precision, scale)
+ | DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok((*precision as i32, *scale as i32))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ &arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Numeric Type for expr",
+ expr,
+ )),
+ }?;
+ let (precision, scale) =
+ round_decimal_base(precision, scale, target_scale, true);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(generic_exec_err(
+ name,
+ "Target scale must be Integer literal, received: None",
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_fields.len()))
+ }?;
+ Ok(return_type)
+}
+
+fn ceil_floor_coerce_first_arg(name: &str, arg_type: &DataType) -> Result<DataType> {
+ if arg_type.is_numeric() {
+ match arg_type {
+ DataType::UInt8 => Ok(DataType::Int16),
+ DataType::UInt16 => Ok(DataType::Int32),
+ DataType::UInt32 | DataType::UInt64 => Ok(DataType::Int64),
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok(DataType::Decimal128(*precision, *scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_type,
+ ))
+ }
+ }
+ other => Ok(other.clone()),
+ }
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ "First arg must be Numeric Type",
+ arg_type,
+ ))
+ }
+}
+
+#[inline]
+fn get_return_type_precision_scale(return_type: &DataType) -> Result<(u8, i8)> {
+ match return_type {
+ DataType::Decimal128(precision, scale) => Ok((*precision, *scale)),
+ other => Err(generic_internal_err(
+ "ceil",
+ format!("Expected return type to be Decimal128, got: {other}").as_str(),
+ )),
+ }
+}
+
+#[derive(Debug)]
+pub struct SparkCeil {
+ signature: Signature,
+}
+
+impl Default for SparkCeil {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkCeil {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkCeil {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn name(&self) -> &str {
+ "ceil"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Err(generic_internal_err(
+ "ceil",
+ "`return_type` should not be called, call `return_type_from_args` instead",
+ ))
+ }
Review Comment:
@irenjj Would `return_field_from_args` work here?
https://github.com/apache/datafusion/blob/7e898620461a4580422a63688351325e3eeb7e89/datafusion/expr/src/udf.rs#L447-L487
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#discussion_r2079751198
##########
datafusion/spark/src/function/math/ceil_floor.rs:
##########
@@ -0,0 +1,720 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use arrow::{
+ array::{ArrayRef, ArrowNativeTypeOp, AsArray},
+ datatypes::{
+ DataType, Decimal128Type, Field, Float32Type, Float64Type, Int16Type, Int32Type,
+ Int64Type, Int8Type, DECIMAL128_MAX_PRECISION, DECIMAL128_MAX_SCALE,
+ },
+};
+use datafusion_common::{Result, ScalarValue};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+use num::integer::{div_ceil, div_floor};
+
+use crate::function::error_utils::{
+ generic_exec_err, generic_internal_err, invalid_arg_count_exec_err,
+ unsupported_data_type_exec_err, unsupported_data_types_exec_err,
+};
+
+use super::round_decimal_base;
+
+fn ceil_floor_coerce_types(name: &str, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ if arg_types.len() == 1 {
+ if arg_types[0].is_numeric() {
+ Ok(vec![ceil_floor_coerce_first_arg(name, &arg_types[0])?])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type",
+ arg_types,
+ ))
+ }
+ } else if arg_types.len() == 2 {
+ if arg_types[0].is_numeric() && arg_types[1].is_integer() {
+ Ok(vec![
+ ceil_floor_coerce_first_arg(name, &arg_types[0])?,
+ DataType::Int32,
+ ])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type for expr and Integer Type for target scale",
+ arg_types,
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_types.len()))
+ }
+}
+
+fn ceil_floor_return_field_from_args(name: &str, args: ReturnFieldArgs) -> Result<Field> {
+ let arg_fields = args.arg_fields;
+ let scalar_arguments = args.scalar_arguments;
+ let return_type = if arg_fields.len() == 1 {
+ match &arg_fields[0].data_type() {
+ DataType::Decimal128(precision, scale) => {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, true);
+ Ok(DataType::Decimal128(precision, scale))
+ }
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, false);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Ok(DataType::Int64),
+ }
+ } else if arg_fields.len() == 2 {
+ if let Some(target_scale) = scalar_arguments[1] {
+ let expr = &arg_fields[0].data_type();
+ let target_scale: i32 = match target_scale {
+ ScalarValue::Int8(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int16(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int32(Some(v)) => Ok(*v),
+ ScalarValue::Int64(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt8(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt16(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt32(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt64(Some(v)) => Ok(*v as i32),
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Target scale must be Integer literal",
+ &target_scale.data_type(),
+ )),
+ }?;
+ if target_scale < -38 {
+ return Err(generic_exec_err(
+ name,
+ "Target scale must be greater than -38",
+ ));
+ }
+ let (precision, scale) = match expr {
+ DataType::Int8 => Ok((3, 0)),
+ DataType::UInt8 | DataType::Int16 => Ok((5, 0)),
+ DataType::UInt16 | DataType::Int32 => Ok((10, 0)),
+ DataType::UInt32 | DataType::UInt64 | DataType::Int64 => Ok((20, 0)),
+ DataType::Float32 => Ok((14, 7)),
+ DataType::Float64 => Ok((30, 15)),
+ DataType::Decimal128(precision, scale)
+ | DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok((*precision as i32, *scale as i32))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Numeric Type for expr",
+ expr,
+ )),
+ }?;
+ let (precision, scale) =
+ round_decimal_base(precision, scale, target_scale, true);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(generic_exec_err(
+ name,
+ "Target scale must be Integer literal, received: None",
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_fields.len()))
+ }?;
+ Ok(Field::new(name.to_string(), return_type, true))
+}
+
+fn ceil_floor_coerce_first_arg(name: &str, arg_type: &DataType) -> Result<DataType> {
+ if arg_type.is_numeric() {
+ match arg_type {
+ DataType::UInt8 => Ok(DataType::Int16),
+ DataType::UInt16 => Ok(DataType::Int32),
+ DataType::UInt32 | DataType::UInt64 => Ok(DataType::Int64),
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok(DataType::Decimal128(*precision, *scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_type,
+ ))
+ }
+ }
+ other => Ok(other.clone()),
+ }
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ "First arg must be Numeric Type",
+ arg_type,
+ ))
+ }
+}
+
+#[inline]
+fn get_return_type_precision_scale(return_type: &DataType) -> Result<(u8, i8)> {
+ match return_type {
+ DataType::Decimal128(precision, scale) => Ok((*precision, *scale)),
+ other => Err(generic_internal_err(
+ "ceil",
+ format!("Expected return type to be Decimal128, got: {other}").as_str(),
+ )),
+ }
+}
+
+#[derive(Debug)]
+pub struct SparkCeil {
+ signature: Signature,
+}
+
+impl Default for SparkCeil {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkCeil {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkCeil {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn name(&self) -> &str {
+ "ceil"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Err(generic_internal_err(
+ "ceil",
+ "`return_type` should not be called, call `return_type_from_args` instead",
+ ))
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<Field> {
+ ceil_floor_return_field_from_args("ceil", args)
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ let arg_len = args.args.len();
+ let target_scale = if arg_len == 1 {
+ Ok(&None)
+ } else if arg_len == 2 {
Review Comment:
I think it is fine to have both cases covered in a single expression in DataFusion. Sorry for the confusion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#discussion_r2078776570
##########
datafusion/sqllogictest/test_files/spark/math/ceil.slt:
##########
@@ -0,0 +1,141 @@
+# 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.
+
+query R
+SELECT ceil(-0.1);
+----
+0
+
+query R
+SELECT ceil(5.4);
+----
+6
+
+query error
+SELECT ceil(3.1411, -3);
+----
+DataFusion error: Error during planning: Failed to coerce arguments to satisfy a call to 'ceil' function: coercion from [Float64, Int64] to the signature Uniform(1, [Float64, Float32]) failed
Review Comment:
I assume that these errors need to be fixed still
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement ceil&floor function for spark [datafusion]
Posted by "irenjj (via GitHub)" <gi...@apache.org>.
irenjj commented on code in PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#discussion_r2075029879
##########
datafusion/spark/src/function/math/ceil_floor.rs:
##########
@@ -0,0 +1,716 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use arrow::{
+ array::{ArrayRef, ArrowNativeTypeOp, AsArray},
+ datatypes::{
+ DataType, Decimal128Type, Float32Type, Float64Type, Int16Type, Int32Type,
+ Int64Type, Int8Type, DECIMAL128_MAX_PRECISION, DECIMAL128_MAX_SCALE,
+ },
+};
+use datafusion_common::{Result, ScalarValue};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+use num::integer::{div_ceil, div_floor};
+
+use crate::function::error_utils::{
+ generic_exec_err, generic_internal_err, invalid_arg_count_exec_err,
+ unsupported_data_type_exec_err, unsupported_data_types_exec_err,
+};
+
+use super::round_decimal_base;
+
+fn ceil_floor_coerce_types(name: &str, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ if arg_types.len() == 1 {
+ if arg_types[0].is_numeric() {
+ Ok(vec![ceil_floor_coerce_first_arg(name, &arg_types[0])?])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type",
+ arg_types,
+ ))
+ }
+ } else if arg_types.len() == 2 {
+ if arg_types[0].is_numeric() && arg_types[1].is_integer() {
+ Ok(vec![
+ ceil_floor_coerce_first_arg(name, &arg_types[0])?,
+ DataType::Int32,
+ ])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type for expr and Integer Type for target scale",
+ arg_types,
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_types.len()))
+ }
+}
+
+#[allow(dead_code)]
+fn ceil_floor_return_type_from_args(
+ name: &str,
+ args: ReturnFieldArgs,
+) -> Result<DataType> {
+ let arg_fields = args.arg_fields;
+ let scalar_arguments = args.scalar_arguments;
+ let return_type = if arg_fields.len() == 1 {
+ match &arg_fields[0].data_type() {
+ DataType::Decimal128(precision, scale) => {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, true);
+ Ok(DataType::Decimal128(precision, scale))
+ }
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, false);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ &arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Ok(DataType::Int64),
+ }
+ } else if arg_fields.len() == 2 {
+ if let Some(target_scale) = scalar_arguments[1] {
+ let expr = &arg_fields[0].data_type();
+ let target_scale: i32 = match target_scale {
+ ScalarValue::Int8(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int16(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int32(Some(v)) => Ok(*v),
+ ScalarValue::Int64(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt8(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt16(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt32(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt64(Some(v)) => Ok(*v as i32),
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Target scale must be Integer literal",
+ &target_scale.data_type(),
+ )),
+ }?;
+ if target_scale < -38 {
+ return Err(generic_exec_err(
+ name,
+ "Target scale must be greater than -38",
+ ));
+ }
+ let (precision, scale) = match expr {
+ DataType::Int8 => Ok((3, 0)),
+ DataType::UInt8 | DataType::Int16 => Ok((5, 0)),
+ DataType::UInt16 | DataType::Int32 => Ok((10, 0)),
+ DataType::UInt32 | DataType::UInt64 | DataType::Int64 => Ok((20, 0)),
+ DataType::Float32 => Ok((14, 7)),
+ DataType::Float64 => Ok((30, 15)),
+ DataType::Decimal128(precision, scale)
+ | DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok((*precision as i32, *scale as i32))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ &arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Numeric Type for expr",
+ expr,
+ )),
+ }?;
+ let (precision, scale) =
+ round_decimal_base(precision, scale, target_scale, true);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(generic_exec_err(
+ name,
+ "Target scale must be Integer literal, received: None",
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_fields.len()))
+ }?;
+ Ok(return_type)
+}
+
+fn ceil_floor_coerce_first_arg(name: &str, arg_type: &DataType) -> Result<DataType> {
+ if arg_type.is_numeric() {
+ match arg_type {
+ DataType::UInt8 => Ok(DataType::Int16),
+ DataType::UInt16 => Ok(DataType::Int32),
+ DataType::UInt32 | DataType::UInt64 => Ok(DataType::Int64),
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok(DataType::Decimal128(*precision, *scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_type,
+ ))
+ }
+ }
+ other => Ok(other.clone()),
+ }
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ "First arg must be Numeric Type",
+ arg_type,
+ ))
+ }
+}
+
+#[inline]
+fn get_return_type_precision_scale(return_type: &DataType) -> Result<(u8, i8)> {
+ match return_type {
+ DataType::Decimal128(precision, scale) => Ok((*precision, *scale)),
+ other => Err(generic_internal_err(
+ "ceil",
+ format!("Expected return type to be Decimal128, got: {other}").as_str(),
+ )),
+ }
+}
+
+#[derive(Debug)]
+pub struct SparkCeil {
+ signature: Signature,
+}
+
+impl Default for SparkCeil {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkCeil {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkCeil {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn name(&self) -> &str {
+ "ceil"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Err(generic_internal_err(
+ "ceil",
+ "`return_type` should not be called, call `return_type_from_args` instead",
+ ))
+ }
Review Comment:
Hi @shehabgamin, we encountered an issue here: most of the code was copied from `sail`. In `sail`, the return value is constructed using `return_type_from_args`, which requires access to both `args.arg_types` and `args.scalar_arguments` to determine the final value. However, in the latest version of DF, `return_type_from_args` has been removed. It seems that relying solely on `datatype` is no longer sufficient to construct the return value within `return_type()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement ceil&floor function for spark [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#issuecomment-2860674623
Hi @irenjj. Could you explain in the PR description how these functions differ from the standard DataFusion implementation (i.e., what is Spark-specific about them)? That will help with reviews.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#issuecomment-2983591412
> > @irenjj - I wonder if you would be willing to help pick this PR back up now that we have merged a PR with a bunch of tests from @shehabgamin here:
> >
> > * [chore: generate basic spark function tests #16409](https://github.com/apache/datafusion/pull/16409)
> >
> > We are hoping to generate some example PRs to use when starting to fill out the rest of the functions and I was thinking this one might be a good one
>
> Sure! I need some time to figure out why new added ceil/floor only works with one argument(it seems datafusion has an original implementation for ceil/floor), i'll do this work next week(little busy this week) -- Sorry for dragging out this PR for so long @shehabgamin @andygrove
Thanks @irenjj -- no worries about the timing. I know you are busy and are also on the leading edge of these spark functions (so the process of implementation is a bit rough). Thank you for helping us figure it out
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#issuecomment-3198977453
Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #15958: [datafusion-spark] Implement ceil&floor function for spark
URL: https://github.com/apache/datafusion/pull/15958
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#discussion_r2078772494
##########
datafusion/spark/src/function/math/ceil_floor.rs:
##########
@@ -0,0 +1,720 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use arrow::{
+ array::{ArrayRef, ArrowNativeTypeOp, AsArray},
+ datatypes::{
+ DataType, Decimal128Type, Field, Float32Type, Float64Type, Int16Type, Int32Type,
+ Int64Type, Int8Type, DECIMAL128_MAX_PRECISION, DECIMAL128_MAX_SCALE,
+ },
+};
+use datafusion_common::{Result, ScalarValue};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+use num::integer::{div_ceil, div_floor};
+
+use crate::function::error_utils::{
+ generic_exec_err, generic_internal_err, invalid_arg_count_exec_err,
+ unsupported_data_type_exec_err, unsupported_data_types_exec_err,
+};
+
+use super::round_decimal_base;
+
+fn ceil_floor_coerce_types(name: &str, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ if arg_types.len() == 1 {
+ if arg_types[0].is_numeric() {
+ Ok(vec![ceil_floor_coerce_first_arg(name, &arg_types[0])?])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type",
+ arg_types,
+ ))
+ }
+ } else if arg_types.len() == 2 {
+ if arg_types[0].is_numeric() && arg_types[1].is_integer() {
+ Ok(vec![
+ ceil_floor_coerce_first_arg(name, &arg_types[0])?,
+ DataType::Int32,
+ ])
+ } else {
+ Err(unsupported_data_types_exec_err(
+ name,
+ "Numeric Type for expr and Integer Type for target scale",
+ arg_types,
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_types.len()))
+ }
+}
+
+fn ceil_floor_return_field_from_args(name: &str, args: ReturnFieldArgs) -> Result<Field> {
+ let arg_fields = args.arg_fields;
+ let scalar_arguments = args.scalar_arguments;
+ let return_type = if arg_fields.len() == 1 {
+ match &arg_fields[0].data_type() {
+ DataType::Decimal128(precision, scale) => {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, true);
+ Ok(DataType::Decimal128(precision, scale))
+ }
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ let (precision, scale) =
+ round_decimal_base(*precision as i32, *scale as i32, 0, false);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Ok(DataType::Int64),
+ }
+ } else if arg_fields.len() == 2 {
+ if let Some(target_scale) = scalar_arguments[1] {
+ let expr = &arg_fields[0].data_type();
+ let target_scale: i32 = match target_scale {
+ ScalarValue::Int8(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int16(Some(v)) => Ok(*v as i32),
+ ScalarValue::Int32(Some(v)) => Ok(*v),
+ ScalarValue::Int64(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt8(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt16(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt32(Some(v)) => Ok(*v as i32),
+ ScalarValue::UInt64(Some(v)) => Ok(*v as i32),
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Target scale must be Integer literal",
+ &target_scale.data_type(),
+ )),
+ }?;
+ if target_scale < -38 {
+ return Err(generic_exec_err(
+ name,
+ "Target scale must be greater than -38",
+ ));
+ }
+ let (precision, scale) = match expr {
+ DataType::Int8 => Ok((3, 0)),
+ DataType::UInt8 | DataType::Int16 => Ok((5, 0)),
+ DataType::UInt16 | DataType::Int32 => Ok((10, 0)),
+ DataType::UInt32 | DataType::UInt64 | DataType::Int64 => Ok((20, 0)),
+ DataType::Float32 => Ok((14, 7)),
+ DataType::Float64 => Ok((30, 15)),
+ DataType::Decimal128(precision, scale)
+ | DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok((*precision as i32, *scale as i32))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_fields[0].data_type(),
+ ))
+ }
+ }
+ _ => Err(unsupported_data_type_exec_err(
+ name,
+ "Numeric Type for expr",
+ expr,
+ )),
+ }?;
+ let (precision, scale) =
+ round_decimal_base(precision, scale, target_scale, true);
+ Ok(DataType::Decimal128(precision, scale))
+ } else {
+ Err(generic_exec_err(
+ name,
+ "Target scale must be Integer literal, received: None",
+ ))
+ }
+ } else {
+ Err(invalid_arg_count_exec_err(name, (1, 2), arg_fields.len()))
+ }?;
+ Ok(Field::new(name.to_string(), return_type, true))
+}
+
+fn ceil_floor_coerce_first_arg(name: &str, arg_type: &DataType) -> Result<DataType> {
+ if arg_type.is_numeric() {
+ match arg_type {
+ DataType::UInt8 => Ok(DataType::Int16),
+ DataType::UInt16 => Ok(DataType::Int32),
+ DataType::UInt32 | DataType::UInt64 => Ok(DataType::Int64),
+ DataType::Decimal256(precision, scale) => {
+ if *precision <= DECIMAL128_MAX_PRECISION
+ && *scale <= DECIMAL128_MAX_SCALE
+ {
+ Ok(DataType::Decimal128(*precision, *scale))
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ format!("Decimal Type must have precision <= {DECIMAL128_MAX_PRECISION} and scale <= {DECIMAL128_MAX_SCALE}").as_str(),
+ arg_type,
+ ))
+ }
+ }
+ other => Ok(other.clone()),
+ }
+ } else {
+ Err(unsupported_data_type_exec_err(
+ name,
+ "First arg must be Numeric Type",
+ arg_type,
+ ))
+ }
+}
+
+#[inline]
+fn get_return_type_precision_scale(return_type: &DataType) -> Result<(u8, i8)> {
+ match return_type {
+ DataType::Decimal128(precision, scale) => Ok((*precision, *scale)),
+ other => Err(generic_internal_err(
+ "ceil",
+ format!("Expected return type to be Decimal128, got: {other}").as_str(),
+ )),
+ }
+}
+
+#[derive(Debug)]
+pub struct SparkCeil {
+ signature: Signature,
+}
+
+impl Default for SparkCeil {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkCeil {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkCeil {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn name(&self) -> &str {
+ "ceil"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Err(generic_internal_err(
+ "ceil",
+ "`return_type` should not be called, call `return_type_from_args` instead",
+ ))
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<Field> {
+ ceil_floor_return_field_from_args("ceil", args)
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ let arg_len = args.args.len();
+ let target_scale = if arg_len == 1 {
+ Ok(&None)
+ } else if arg_len == 2 {
Review Comment:
Actually, I have confused myself now. The plan does still show two arguments, but the physical plan only has a single argument :confused:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "shehabgamin (via GitHub)" <gi...@apache.org>.
shehabgamin commented on code in PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#discussion_r2080822728
##########
datafusion/sqllogictest/test_files/spark/math/ceil.slt:
##########
@@ -0,0 +1,141 @@
+# 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.
+
+query R
+SELECT ceil(-0.1);
+----
+0
+
+query R
+SELECT ceil(5.4);
+----
+6
+
+query error
+SELECT ceil(3.1411, -3);
+----
+DataFusion error: Error during planning: Failed to coerce arguments to satisfy a call to 'ceil' function: coercion from [Float64, Int64] to the signature Uniform(1, [Float64, Float32]) failed
Review Comment:
@irenjj Sail doesn't use the same SQL parser as DataFusion, as it follows Spark SQL syntax.
You’ll need to cast values to ensure the tests interpret the correct `DataType`.
For example, instead of this:
```
SELECT ceil(3.1411, -3);
```
Use:
```
SELECT ceil(3.1411::decimal(5,4), -3::INT);
```
Copy-pasting the exact queries from the files below won't work directly. However, if you look closely, you’ll see that each query includes the necessary cast for the first argument (the second argument is always an `INT`):
- https://github.com/lakehq/sail/blob/main/python/pysail/tests/spark/function/test_ceil.txt
- https://github.com/lakehq/sail/blob/main/python/pysail/tests/spark/function/test_floor.txt
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement ceil&floor function for spark [datafusion]
Posted by "shehabgamin (via GitHub)" <gi...@apache.org>.
shehabgamin commented on PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#issuecomment-2860636654
@irenjj I'll review this by tomorrow!
I'm not a committer though, so we'll still need a review from someone else as well. cc @alamb @andygrove
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-spark] Implement ceil&floor function for spark [datafusion]
Posted by "irenjj (via GitHub)" <gi...@apache.org>.
irenjj commented on PR #15958:
URL: https://github.com/apache/datafusion/pull/15958#issuecomment-2982098992
> @irenjj - I wonder if you would be willing to help pick this PR back up now that we have merged a PR with a bunch of tests from @shehabgamin here:
>
> * [chore: generate basic spark function tests #16409](https://github.com/apache/datafusion/pull/16409)
>
> We are hoping to generate some example PRs to use when starting to fill out the rest of the functions and I was thinking this one might be a good one
Sure! I need some time to figure out why new added ceil/floor only works with one argument(it seems datafusion has an original implementation for ceil/floor), i'll do this work next week(little busy this week) -- Sorry for dragging out this PR for so long @shehabgamin @andygrove
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org