[pull] main from apache:main

datafusion/functions/Cargo.toml

@@ -98,6 +98,11 @@ env_logger = { workspace = true }
 rand = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt", "sync"] }
 
+[[bench]]
+harness = false
+name = "round_dense"
+required-features = ["math_expressions"]
+
 [[bench]]
 harness = false
 name = "ascii"

datafusion/functions/benches/date_trunc.rs

@@ -18,52 +18,64 @@
 use std::hint::black_box;
 use std::sync::Arc;
 
-use arrow::array::{Array, ArrayRef, TimestampSecondArray};
+use arrow::array::{Array, ArrayRef, TimestampNanosecondArray, TimestampSecondArray};
 use arrow::datatypes::Field;
 use criterion::{Criterion, criterion_group, criterion_main};
 use datafusion_common::ScalarValue;
 use datafusion_common::config::ConfigOptions;
 use datafusion_expr::{ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs};
 use datafusion_functions::datetime::date_trunc;
-use rand::Rng;
-use rand::rngs::ThreadRng;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
 
-fn timestamps(rng: &mut ThreadRng) -> TimestampSecondArray {
-    let mut seconds = vec![];
-    for _ in 0..1000 {
-        seconds.push(rng.random_range(0..1_000_000));
-    }
+const NUM_ROWS: usize = 1000;
+const NANOS_PER_SECOND: i64 = 1_000_000_000;
+/// Roughly 30 years, so that values span many months, quarters and years.
+const RANGE_SECONDS: i64 = 30 * 365 * 24 * 60 * 60;
 
-    TimestampSecondArray::from(seconds)
+fn seedable_rng() -> StdRng {
+    StdRng::seed_from_u64(42)
 }
 
-fn criterion_benchmark(c: &mut Criterion) {
-    c.bench_function("date_trunc_minute_1000", |b| {
-        let mut rng = rand::rng();
-        let timestamps_array = Arc::new(timestamps(&mut rng)) as ArrayRef;
-        let batch_len = timestamps_array.len();
-        let precision =
-            ColumnarValue::Scalar(ScalarValue::Utf8(Some("minute".to_string())));
-        let timestamps = ColumnarValue::Array(timestamps_array);
-        let udf = date_trunc();
-        let args = vec![precision, timestamps];
-        let arg_fields = args
-            .iter()
-            .enumerate()
-            .map(|(idx, arg)| {
-                Field::new(format!("arg_{idx}"), arg.data_type(), true).into()
-            })
-            .collect::<Vec<_>>();
+fn second_timestamps() -> TimestampSecondArray {
+    let mut rng = seedable_rng();
+    (0..NUM_ROWS)
+        .map(|_| Some(rng.random_range(0..1_000_000i64)))
+        .collect()
+}
 
-        let scalar_arguments = vec![None; arg_fields.len()];
-        let return_field = udf
-            .return_field_from_args(ReturnFieldArgs {
-                arg_fields: &arg_fields,
-                scalar_arguments: &scalar_arguments,
-            })
-            .unwrap();
-        let config_options = Arc::new(ConfigOptions::default());
+fn nanosecond_timestamps() -> TimestampNanosecondArray {
+    let mut rng = seedable_rng();
+    (0..NUM_ROWS)
+        .map(|_| {
+            let seconds = rng.random_range(-RANGE_SECONDS..RANGE_SECONDS);
+            Some(seconds * NANOS_PER_SECOND + rng.random_range(0..NANOS_PER_SECOND))
+        })
+        .collect()
+}
+
+fn run_benchmark(c: &mut Criterion, name: &str, granularity: &str, array: ArrayRef) {
+    let batch_len = array.len();
+    let precision =
+        ColumnarValue::Scalar(ScalarValue::Utf8(Some(granularity.to_string())));
+    let udf = date_trunc();
+    let args = vec![precision, ColumnarValue::Array(array)];
+    let arg_fields = args
+        .iter()
+        .enumerate()
+        .map(|(idx, arg)| Field::new(format!("arg_{idx}"), arg.data_type(), true).into())
+        .collect::<Vec<_>>();
 
+    let scalar_arguments = vec![None; arg_fields.len()];
+    let return_field = udf
+        .return_field_from_args(ReturnFieldArgs {
+            arg_fields: &arg_fields,
+            scalar_arguments: &scalar_arguments,
+        })
+        .unwrap();
+    let config_options = Arc::new(ConfigOptions::default());
+
+    c.bench_function(name, |b| {
         b.iter(|| {
             black_box(
                 udf.invoke_with_args(ScalarFunctionArgs {
@@ -79,5 +91,23 @@ fn criterion_benchmark(c: &mut Criterion) {
     });
 }
 
+fn criterion_benchmark(c: &mut Criterion) {
+    let seconds: ArrayRef = Arc::new(second_timestamps());
+    run_benchmark(c, "date_trunc_minute_1000", "minute", Arc::clone(&seconds));
+    run_benchmark(c, "date_trunc_month_second_1000", "month", seconds);
+
+    // Coarse granularities on an untimezoned array: these need calendar
+    // arithmetic rather than a plain division.
+    let nanos: ArrayRef = Arc::new(nanosecond_timestamps());
+    for granularity in ["week", "month", "quarter", "year"] {
+        run_benchmark(
+            c,
+            &format!("date_trunc_{granularity}_nanos_1000"),
+            granularity,
+            Arc::clone(&nanos),
+        );
+    }
+}
+
 criterion_group!(benches, criterion_benchmark);
 criterion_main!(benches);

datafusion/functions/benches/round_dense.rs

@@ -0,0 +1,94 @@
+// 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.
+
+//! Microbenchmark for `round(float_array, scalar_decimal_places)` over a
+//! Float column with no NULLs — the dense elementwise-rounding path.
+
+use arrow::array::ArrayRef;
+use arrow::datatypes::{DataType, Field, Float32Type, Float64Type};
+use arrow::util::bench_util::create_primitive_array;
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use datafusion_common::ScalarValue;
+use datafusion_common::config::ConfigOptions;
+use datafusion_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl};
+use datafusion_functions::math::round::RoundFunc;
+use std::hint::black_box;
+use std::sync::Arc;
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let round_fn = RoundFunc::new();
+    let config_options = Arc::new(ConfigOptions::default());
+
+    for size in [1024usize, 4096, 8192] {
+        // Float64, no nulls.
+        let f64_array: ArrayRef =
+            Arc::new(create_primitive_array::<Float64Type>(size, 0.0));
+        let f64_args = vec![
+            ColumnarValue::Array(Arc::clone(&f64_array)),
+            ColumnarValue::Scalar(ScalarValue::Int32(Some(2))),
+        ];
+        c.bench_with_input(BenchmarkId::new("round_dense_f64", size), &size, |b, _| {
+            b.iter(|| {
+                black_box(
+                    round_fn
+                        .invoke_with_args(ScalarFunctionArgs {
+                            args: f64_args.clone(),
+                            arg_fields: vec![
+                                Field::new("a", DataType::Float64, false).into(),
+                                Field::new("b", DataType::Int32, false).into(),
+                            ],
+                            number_rows: size,
+                            return_field: Field::new("f", DataType::Float64, false)
+                                .into(),
+                            config_options: Arc::clone(&config_options),
+                        })
+                        .unwrap(),
+                )
+            })
+        });
+
+        // Float32, no nulls.
+        let f32_array: ArrayRef =
+            Arc::new(create_primitive_array::<Float32Type>(size, 0.0));
+        let f32_args = vec![
+            ColumnarValue::Array(Arc::clone(&f32_array)),
+            ColumnarValue::Scalar(ScalarValue::Int32(Some(2))),
+        ];
+        c.bench_with_input(BenchmarkId::new("round_dense_f32", size), &size, |b, _| {
+            b.iter(|| {
+                black_box(
+                    round_fn
+                        .invoke_with_args(ScalarFunctionArgs {
+                            args: f32_args.clone(),
+                            arg_fields: vec![
+                                Field::new("a", DataType::Float32, false).into(),
+                                Field::new("b", DataType::Int32, false).into(),
+                            ],
+                            number_rows: size,
+                            return_field: Field::new("f", DataType::Float32, false)
+                                .into(),
+                            config_options: Arc::clone(&config_options),
+                        })
+                        .unwrap(),
+                )
+            })
+        });
+    }
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);

datafusion/functions/src/datetime/date_part.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::iter::repeat_n;
 use std::str::FromStr;
 use std::sync::Arc;
 
@@ -390,14 +391,27 @@ fn part_normalization(part: &str) -> &str {
 
 /// Invoke [`date_part`] on an `array` (e.g. Timestamp) and convert the
 /// result to a total number of seconds, milliseconds, microseconds or
-/// nanoseconds
+/// nanoseconds as an `Int32Array`
 fn seconds_as_i32(array: &dyn Array, unit: TimeUnit) -> Result<ArrayRef> {
     // Nanosecond is neither supported in Postgres nor DuckDB, to avoid dealing
     // with overflow and precision issue we don't support nanosecond
     if unit == Nanosecond {
         return not_impl_err!("Date part {unit:?} not supported");
     }
 
+    // Fast path with seconds - no need to compute nanoseconds
+    if unit == Second {
+        return Ok(date_part(array, DatePart::Second)?);
+    }
+
+    // Fast path for Date32 and Date64 - no seconds
+    if array.data_type() == &Date32 || array.data_type() == &Date64 {
+        return Ok(Arc::new(Int32Array::from_iter_values_with_nulls(
+            repeat_n(0, array.len()),
+            array.nulls().cloned(),
+        )));
+    }
+
     let conversion_factor = match unit {
         Second => 1_000_000_000,
         Millisecond => 1_000_000,
@@ -539,6 +553,14 @@ fn epoch(array: &dyn Array) -> Result<ArrayRef> {
 /// `nanosecond`s in each second, so representing up to 60 seconds as
 /// nanoseconds can be values up to 60 billion, which does not fit in Int32.
 fn seconds_ns(array: &dyn Array) -> Result<ArrayRef> {
+    // Fast path for Date32 and Date64 - no nanoseconds
+    if array.data_type() == &Date32 || array.data_type() == &Date64 {
+        return Ok(Arc::new(Int64Array::from_iter_values_with_nulls(
+            repeat_n(0, array.len()),
+            array.nulls().cloned(),
+        )));
+    }
+
     let secs = date_part(array, DatePart::Second)?;
     // This assumes array is primitive and not a dictionary
     let secs = as_int32_array(secs.as_ref())?;