This is an automated email from the ASF dual-hosted git repository.
github-bot pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 4d8d48c0c7 perf: Optimize scalar performance for cot (#19888)
4d8d48c0c7 is described below
commit 4d8d48c0c7bc9db9f8ad475765095ac3f8458e79
Author: Kumar Ujjawal <[email protected]>
AuthorDate: Wed Jan 21 09:14:22 2026 +0530
perf: Optimize scalar performance for cot (#19888)
## 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.
-->
- Part of https://github.com/apache/datafusion-comet/issues/2986.
## Rationale for this change
The cot function currently converts scalar inputs to arrays before
processing, even for single scalar values. This adds unnecessary
overhead from array allocation and conversion. Adding a scalar fast path
avoids this overhead.
<!--
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?
- Added scalar fast path
- Added benchmark
- Update tests
| Type | Before | After | Speedup |
|------|--------|-------|---------|
| **cot_f64_scalar** | 229 ns | 67 ns | **3.4x** |
| **cot_f32_scalar** | 247 ns | 59 ns | **4.2x** |
<!--
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.
-->
---
datafusion/functions/benches/cot.rs | 47 +++++-
datafusion/functions/src/math/cot.rs | 301 ++++++++++++++++++++++++++++-------
2 files changed, 287 insertions(+), 61 deletions(-)
diff --git a/datafusion/functions/benches/cot.rs
b/datafusion/functions/benches/cot.rs
index c47198d4a6..061d14cbf0 100644
--- a/datafusion/functions/benches/cot.rs
+++ b/datafusion/functions/benches/cot.rs
@@ -27,11 +27,15 @@ use datafusion_functions::math::cot;
use std::hint::black_box;
use arrow::datatypes::{DataType, Field};
+use datafusion_common::ScalarValue;
use datafusion_common::config::ConfigOptions;
use std::sync::Arc;
fn criterion_benchmark(c: &mut Criterion) {
let cot_fn = cot();
+ let config_options = Arc::new(ConfigOptions::default());
+
+ // Array benchmarks - run for different sizes
for size in [1024, 4096, 8192] {
let f32_array = Arc::new(create_primitive_array::<Float32Type>(size,
0.2));
let f32_args = vec![ColumnarValue::Array(f32_array)];
@@ -42,7 +46,6 @@ fn criterion_benchmark(c: &mut Criterion) {
Field::new(format!("arg_{idx}"), arg.data_type(), true).into()
})
.collect::<Vec<_>>();
- let config_options = Arc::new(ConfigOptions::default());
c.bench_function(&format!("cot f32 array: {size}"), |b| {
b.iter(|| {
@@ -59,6 +62,7 @@ fn criterion_benchmark(c: &mut Criterion) {
)
})
});
+
let f64_array = Arc::new(create_primitive_array::<Float64Type>(size,
0.2));
let f64_args = vec![ColumnarValue::Array(f64_array)];
let arg_fields = f64_args
@@ -86,6 +90,47 @@ fn criterion_benchmark(c: &mut Criterion) {
})
});
}
+
+ // Scalar benchmarks - run only once since size doesn't affect scalar
performance
+ let scalar_f32_args =
vec![ColumnarValue::Scalar(ScalarValue::Float32(Some(1.0)))];
+ let scalar_f32_arg_fields = vec![Field::new("a", DataType::Float32,
false).into()];
+ let return_field_f32 = Field::new("f", DataType::Float32, false).into();
+
+ c.bench_function("cot f32 scalar", |b| {
+ b.iter(|| {
+ black_box(
+ cot_fn
+ .invoke_with_args(ScalarFunctionArgs {
+ args: scalar_f32_args.clone(),
+ arg_fields: scalar_f32_arg_fields.clone(),
+ number_rows: 1,
+ return_field: Arc::clone(&return_field_f32),
+ config_options: Arc::clone(&config_options),
+ })
+ .unwrap(),
+ )
+ })
+ });
+
+ let scalar_f64_args =
vec![ColumnarValue::Scalar(ScalarValue::Float64(Some(1.0)))];
+ let scalar_f64_arg_fields = vec![Field::new("a", DataType::Float64,
false).into()];
+ let return_field_f64 = Field::new("f", DataType::Float64, false).into();
+
+ c.bench_function("cot f64 scalar", |b| {
+ b.iter(|| {
+ black_box(
+ cot_fn
+ .invoke_with_args(ScalarFunctionArgs {
+ args: scalar_f64_args.clone(),
+ arg_fields: scalar_f64_arg_fields.clone(),
+ number_rows: 1,
+ return_field: Arc::clone(&return_field_f64),
+ config_options: Arc::clone(&config_options),
+ })
+ .unwrap(),
+ )
+ })
+ });
}
criterion_group!(benches, criterion_benchmark);
diff --git a/datafusion/functions/src/math/cot.rs
b/datafusion/functions/src/math/cot.rs
index a0d7b02b68..1f67ef7138 100644
--- a/datafusion/functions/src/math/cot.rs
+++ b/datafusion/functions/src/math/cot.rs
@@ -18,12 +18,12 @@
use std::any::Any;
use std::sync::Arc;
-use arrow::array::{ArrayRef, AsArray};
+use arrow::array::AsArray;
use arrow::datatypes::DataType::{Float32, Float64};
use arrow::datatypes::{DataType, Float32Type, Float64Type};
-use crate::utils::make_scalar_function;
-use datafusion_common::{Result, exec_err};
+use datafusion_common::utils::take_function_args;
+use datafusion_common::{Result, ScalarValue, internal_err};
use datafusion_expr::{ColumnarValue, Documentation, ScalarFunctionArgs};
use datafusion_expr::{ScalarUDFImpl, Signature, Volatility};
use datafusion_macros::user_doc;
@@ -96,24 +96,47 @@ impl ScalarUDFImpl for CotFunc {
}
fn invoke_with_args(&self, args: ScalarFunctionArgs) ->
Result<ColumnarValue> {
- make_scalar_function(cot, vec![])(&args.args)
- }
-}
+ let return_field = args.return_field;
+ let [arg] = take_function_args(self.name(), args.args)?;
+
+ match arg {
+ ColumnarValue::Scalar(scalar) => {
+ if scalar.is_null() {
+ return ColumnarValue::Scalar(ScalarValue::Null)
+ .cast_to(return_field.data_type(), None);
+ }
-///cot SQL function
-fn cot(args: &[ArrayRef]) -> Result<ArrayRef> {
- match args[0].data_type() {
- Float64 => Ok(Arc::new(
- args[0]
- .as_primitive::<Float64Type>()
- .unary::<_, Float64Type>(|x: f64| compute_cot64(x)),
- ) as ArrayRef),
- Float32 => Ok(Arc::new(
- args[0]
- .as_primitive::<Float32Type>()
- .unary::<_, Float32Type>(|x: f32| compute_cot32(x)),
- ) as ArrayRef),
- other => exec_err!("Unsupported data type {other:?} for function cot"),
+ match scalar {
+ ScalarValue::Float64(Some(v)) => Ok(ColumnarValue::Scalar(
+ ScalarValue::Float64(Some(compute_cot64(v))),
+ )),
+ ScalarValue::Float32(Some(v)) => Ok(ColumnarValue::Scalar(
+ ScalarValue::Float32(Some(compute_cot32(v))),
+ )),
+ _ => {
+ internal_err!(
+ "Unexpected scalar type for cot: {:?}",
+ scalar.data_type()
+ )
+ }
+ }
+ }
+ ColumnarValue::Array(array) => match array.data_type() {
+ Float64 => Ok(ColumnarValue::Array(Arc::new(
+ array
+ .as_primitive::<Float64Type>()
+ .unary::<_, Float64Type>(compute_cot64),
+ ))),
+ Float32 => Ok(ColumnarValue::Array(Arc::new(
+ array
+ .as_primitive::<Float32Type>()
+ .unary::<_, Float32Type>(compute_cot32),
+ ))),
+ other => {
+ internal_err!("Unexpected data type {other:?} for function
cot")
+ }
+ },
+ }
}
}
@@ -129,54 +152,212 @@ fn compute_cot64(x: f64) -> f64 {
#[cfg(test)]
mod test {
- use crate::math::cot::cot;
+ use std::sync::Arc;
+
use arrow::array::{ArrayRef, Float32Array, Float64Array};
+ use arrow::datatypes::{DataType, Field};
+ use datafusion_common::ScalarValue;
use datafusion_common::cast::{as_float32_array, as_float64_array};
- use std::sync::Arc;
+ use datafusion_common::config::ConfigOptions;
+ use datafusion_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl};
+
+ use crate::math::cot::CotFunc;
#[test]
fn test_cot_f32() {
- let args: Vec<ArrayRef> =
- vec![Arc::new(Float32Array::from(vec![12.1, 30.0, 90.0, -30.0]))];
- let result = cot(&args).expect("failed to initialize function cot");
- let floats =
- as_float32_array(&result).expect("failed to initialize function
cot");
-
- let expected = Float32Array::from(vec![
- -1.986_460_4,
- -0.156_119_96,
- -0.501_202_8,
- 0.156_119_96,
- ]);
-
- let eps = 1e-6;
- assert_eq!(floats.len(), 4);
- assert!((floats.value(0) - expected.value(0)).abs() < eps);
- assert!((floats.value(1) - expected.value(1)).abs() < eps);
- assert!((floats.value(2) - expected.value(2)).abs() < eps);
- assert!((floats.value(3) - expected.value(3)).abs() < eps);
+ let array = Arc::new(Float32Array::from(vec![12.1, 30.0, 90.0,
-30.0]));
+ let arg_fields = vec![Field::new("a", DataType::Float32,
false).into()];
+ let args = ScalarFunctionArgs {
+ args: vec![ColumnarValue::Array(Arc::clone(&array) as ArrayRef)],
+ arg_fields,
+ number_rows: array.len(),
+ return_field: Field::new("f", DataType::Float32, true).into(),
+ config_options: Arc::new(ConfigOptions::default()),
+ };
+ let result = CotFunc::new()
+ .invoke_with_args(args)
+ .expect("failed to initialize function cot");
+
+ match result {
+ ColumnarValue::Array(arr) => {
+ let floats = as_float32_array(&arr)
+ .expect("failed to convert result to a Float32Array");
+
+ let expected = Float32Array::from(vec![
+ -1.986_460_4,
+ -0.156_119_96,
+ -0.501_202_8,
+ 0.156_119_96,
+ ]);
+
+ let eps = 1e-6;
+ assert_eq!(floats.len(), 4);
+ assert!((floats.value(0) - expected.value(0)).abs() < eps);
+ assert!((floats.value(1) - expected.value(1)).abs() < eps);
+ assert!((floats.value(2) - expected.value(2)).abs() < eps);
+ assert!((floats.value(3) - expected.value(3)).abs() < eps);
+ }
+ ColumnarValue::Scalar(_) => {
+ panic!("Expected an array value")
+ }
+ }
}
#[test]
fn test_cot_f64() {
- let args: Vec<ArrayRef> =
- vec![Arc::new(Float64Array::from(vec![12.1, 30.0, 90.0, -30.0]))];
- let result = cot(&args).expect("failed to initialize function cot");
- let floats =
- as_float64_array(&result).expect("failed to initialize function
cot");
-
- let expected = Float64Array::from(vec![
- -1.986_458_685_881_4,
- -0.156_119_952_161_6,
- -0.501_202_783_380_1,
- 0.156_119_952_161_6,
- ]);
-
- let eps = 1e-12;
- assert_eq!(floats.len(), 4);
- assert!((floats.value(0) - expected.value(0)).abs() < eps);
- assert!((floats.value(1) - expected.value(1)).abs() < eps);
- assert!((floats.value(2) - expected.value(2)).abs() < eps);
- assert!((floats.value(3) - expected.value(3)).abs() < eps);
+ let array = Arc::new(Float64Array::from(vec![12.1, 30.0, 90.0,
-30.0]));
+ let arg_fields = vec![Field::new("a", DataType::Float64,
false).into()];
+ let args = ScalarFunctionArgs {
+ args: vec![ColumnarValue::Array(Arc::clone(&array) as ArrayRef)],
+ arg_fields,
+ number_rows: array.len(),
+ return_field: Field::new("f", DataType::Float64, true).into(),
+ config_options: Arc::new(ConfigOptions::default()),
+ };
+ let result = CotFunc::new()
+ .invoke_with_args(args)
+ .expect("failed to initialize function cot");
+
+ match result {
+ ColumnarValue::Array(arr) => {
+ let floats = as_float64_array(&arr)
+ .expect("failed to convert result to a Float64Array");
+
+ let expected = Float64Array::from(vec![
+ -1.986_458_685_881_4,
+ -0.156_119_952_161_6,
+ -0.501_202_783_380_1,
+ 0.156_119_952_161_6,
+ ]);
+
+ let eps = 1e-12;
+ assert_eq!(floats.len(), 4);
+ assert!((floats.value(0) - expected.value(0)).abs() < eps);
+ assert!((floats.value(1) - expected.value(1)).abs() < eps);
+ assert!((floats.value(2) - expected.value(2)).abs() < eps);
+ assert!((floats.value(3) - expected.value(3)).abs() < eps);
+ }
+ ColumnarValue::Scalar(_) => {
+ panic!("Expected an array value")
+ }
+ }
+ }
+
+ #[test]
+ fn test_cot_scalar_f64() {
+ let arg_fields = vec![Field::new("a", DataType::Float64,
false).into()];
+ let args = ScalarFunctionArgs {
+ args: vec![ColumnarValue::Scalar(ScalarValue::Float64(Some(1.0)))],
+ arg_fields,
+ number_rows: 1,
+ return_field: Field::new("f", DataType::Float64, false).into(),
+ config_options: Arc::new(ConfigOptions::default()),
+ };
+ let result = CotFunc::new()
+ .invoke_with_args(args)
+ .expect("cot scalar should succeed");
+
+ match result {
+ ColumnarValue::Scalar(ScalarValue::Float64(Some(v))) => {
+ // cot(1.0) = 1/tan(1.0) ≈ 0.6420926159343306
+ let expected = 1.0_f64 / 1.0_f64.tan();
+ assert!((v - expected).abs() < 1e-12);
+ }
+ _ => panic!("Expected Float64 scalar"),
+ }
+ }
+
+ #[test]
+ fn test_cot_scalar_f32() {
+ let arg_fields = vec![Field::new("a", DataType::Float32,
false).into()];
+ let args = ScalarFunctionArgs {
+ args: vec![ColumnarValue::Scalar(ScalarValue::Float32(Some(1.0)))],
+ arg_fields,
+ number_rows: 1,
+ return_field: Field::new("f", DataType::Float32, false).into(),
+ config_options: Arc::new(ConfigOptions::default()),
+ };
+ let result = CotFunc::new()
+ .invoke_with_args(args)
+ .expect("cot scalar should succeed");
+
+ match result {
+ ColumnarValue::Scalar(ScalarValue::Float32(Some(v))) => {
+ let expected = 1.0_f32 / 1.0_f32.tan();
+ assert!((v - expected).abs() < 1e-6);
+ }
+ _ => panic!("Expected Float32 scalar"),
+ }
+ }
+
+ #[test]
+ fn test_cot_scalar_null() {
+ let arg_fields = vec![Field::new("a", DataType::Float64, true).into()];
+ let args = ScalarFunctionArgs {
+ args: vec![ColumnarValue::Scalar(ScalarValue::Float64(None))],
+ arg_fields,
+ number_rows: 1,
+ return_field: Field::new("f", DataType::Float64, true).into(),
+ config_options: Arc::new(ConfigOptions::default()),
+ };
+ let result = CotFunc::new()
+ .invoke_with_args(args)
+ .expect("cot null should succeed");
+
+ match result {
+ ColumnarValue::Scalar(scalar) => {
+ assert!(scalar.is_null());
+ }
+ _ => panic!("Expected scalar result"),
+ }
+ }
+
+ #[test]
+ fn test_cot_scalar_zero() {
+ let arg_fields = vec![Field::new("a", DataType::Float64,
false).into()];
+ let args = ScalarFunctionArgs {
+ args: vec![ColumnarValue::Scalar(ScalarValue::Float64(Some(0.0)))],
+ arg_fields,
+ number_rows: 1,
+ return_field: Field::new("f", DataType::Float64, false).into(),
+ config_options: Arc::new(ConfigOptions::default()),
+ };
+ let result = CotFunc::new()
+ .invoke_with_args(args)
+ .expect("cot zero should succeed");
+
+ match result {
+ ColumnarValue::Scalar(ScalarValue::Float64(Some(v))) => {
+ // cot(0) = 1/tan(0) = infinity
+ assert!(v.is_infinite());
+ }
+ _ => panic!("Expected Float64 scalar"),
+ }
+ }
+
+ #[test]
+ fn test_cot_scalar_pi() {
+ let arg_fields = vec![Field::new("a", DataType::Float64,
false).into()];
+ let args = ScalarFunctionArgs {
+ args: vec![ColumnarValue::Scalar(ScalarValue::Float64(Some(
+ std::f64::consts::PI,
+ )))],
+ arg_fields,
+ number_rows: 1,
+ return_field: Field::new("f", DataType::Float64, false).into(),
+ config_options: Arc::new(ConfigOptions::default()),
+ };
+ let result = CotFunc::new()
+ .invoke_with_args(args)
+ .expect("cot pi should succeed");
+
+ match result {
+ ColumnarValue::Scalar(ScalarValue::Float64(Some(v))) => {
+ // cot(PI) = 1/tan(PI) - very large negative number due to
floating point
+ let expected = 1.0_f64 / std::f64::consts::PI.tan();
+ assert!((v - expected).abs() < 1e-6);
+ }
+ _ => panic!("Expected Float64 scalar"),
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
