perf: Optimize lpad(), rpad() for scalar args (#20657)

## Which issue does this PR close?

- Closes #20655.

## Rationale for this change

`lpad` and `rpad` are commonly called with scalar length and fill
arguments, e.g. `lpad(column, 20, '0')`. We can special-case this
scenario to improve performance by avoiding the overhead of
`make_scalar_function` and by precomputing the padding buffer and
reusing it for each row.

For scalar args, this improves performance by ~3x for ASCII inputs and
~1.6x for Unicode inputs.

## What changes are included in this PR?

- Add benchmarks for padding with scalar length and fill.
- Add a scalar fast path for `lpad` and `rpad` that precomputes a
padding buffer. We only use the fast path if the pad length is
reasonably small (<= 16KB), to avoid using too much memory on a scratch
buffer.
- Code cleanup: extract and use `try_as_scalar_str` and
`try_as_scalar_i64` helpers.
- Code cleanup: use `target_len` consistently instead of `length`,
because the latter is ambiguous.
- Code cleanup: make `rpad` and `lpad` more similar by removing needless
and probably unintended differences between the two functions. We could
go further and refactor them to remove the redundancy but I won't
attempt that for now.

## Are these changes tested?

Yes; covered by existing tests. Added new benchmarks.

## Are there any user-facing changes?

No.

## AI usage

Multiple AI tools were used to iterate on this PR. I have reviewed and
understand the resulting code.

Author: neilconway

datafusion/functions/benches/pad.rs

@@ -24,6 +24,7 @@ use arrow::util::bench_util::{
     create_string_array_with_len, create_string_view_array_with_len,
 };
 use criterion::{Criterion, SamplingMode, criterion_group, criterion_main};
+use datafusion_common::ScalarValue;
 use datafusion_common::config::ConfigOptions;
 use datafusion_expr::{ColumnarValue, ScalarFunctionArgs};
 use datafusion_functions::unicode;
@@ -172,6 +173,32 @@ fn create_pad_args<O: OffsetSizeTrait>(
     }
 }
 
+/// Create args for pad benchmark with scalar length and fill (common pattern:
+/// `lpad(column, 20, '0')`).
+fn create_scalar_pad_args<O: OffsetSizeTrait>(
+    size: usize,
+    str_len: usize,
+    target_len: i64,
+    fill: &str,
+    use_string_view: bool,
+) -> Vec<ColumnarValue> {
+    if use_string_view {
+        let string_array = create_string_view_array_with_len(size, 0.1, str_len, false);
+        vec![
+            ColumnarValue::Array(Arc::new(string_array)),
+            ColumnarValue::Scalar(ScalarValue::Int64(Some(target_len))),
+            ColumnarValue::Scalar(ScalarValue::Utf8(Some(fill.to_string()))),
+        ]
+    } else {
+        let string_array = create_string_array_with_len::<O>(size, 0.1, str_len);
+        vec![
+            ColumnarValue::Array(Arc::new(string_array)),
+            ColumnarValue::Scalar(ScalarValue::Int64(Some(target_len))),
+            ColumnarValue::Scalar(ScalarValue::Utf8(Some(fill.to_string()))),
+        ]
+    }
+}
+
 fn criterion_benchmark(c: &mut Criterion) {
     for size in [1024, 4096] {
         let mut group = c.benchmark_group(format!("lpad size={size}"));
@@ -336,6 +363,118 @@ fn criterion_benchmark(c: &mut Criterion) {
             },
         );
 
+        // --- Scalar length + fill benchmarks ---
+
+        // Utf8 with scalar length and fill (3-arg)
+        let args = create_scalar_pad_args::<i32>(size, 5, 20, "x", false);
+        let arg_fields = args
+            .iter()
+            .enumerate()
+            .map(|(idx, arg)| {
+                Field::new(format!("arg_{idx}"), arg.data_type(), true).into()
+            })
+            .collect::<Vec<_>>();
+
+        group.bench_function(
+            format!("lpad utf8 scalar [size={size}, str_len=5, target=20, fill='x']"),
+            |b| {
+                b.iter(|| {
+                    let args_cloned = args.clone();
+                    black_box(unicode::lpad().invoke_with_args(ScalarFunctionArgs {
+                        args: args_cloned,
+                        arg_fields: arg_fields.clone(),
+                        number_rows: size,
+                        return_field: Field::new("f", DataType::Utf8, true).into(),
+                        config_options: Arc::clone(&config_options),
+                    }))
+                })
+            },
+        );
+
+        // StringView with scalar length and fill (3-arg)
+        let args = create_scalar_pad_args::<i32>(size, 5, 20, "x", true);
+        let arg_fields = args
+            .iter()
+            .enumerate()
+            .map(|(idx, arg)| {
+                Field::new(format!("arg_{idx}"), arg.data_type(), true).into()
+            })
+            .collect::<Vec<_>>();
+
+        group.bench_function(
+            format!(
+                "lpad stringview scalar [size={size}, str_len=5, target=20, fill='x']"
+            ),
+            |b| {
+                b.iter(|| {
+                    let args_cloned = args.clone();
+                    black_box(unicode::lpad().invoke_with_args(ScalarFunctionArgs {
+                        args: args_cloned,
+                        arg_fields: arg_fields.clone(),
+                        number_rows: size,
+                        return_field: Field::new("f", DataType::Utf8, true).into(),
+                        config_options: Arc::clone(&config_options),
+                    }))
+                })
+            },
+        );
+
+        // Utf8 with scalar length and unicode fill
+        let args = create_scalar_pad_args::<i32>(size, 5, 20, "é", false);
+        let arg_fields = args
+            .iter()
+            .enumerate()
+            .map(|(idx, arg)| {
+                Field::new(format!("arg_{idx}"), arg.data_type(), true).into()
+            })
+            .collect::<Vec<_>>();
+
+        group.bench_function(
+            format!(
+                "lpad utf8 scalar unicode [size={size}, str_len=5, target=20, fill='é']"
+            ),
+            |b| {
+                b.iter(|| {
+                    let args_cloned = args.clone();
+                    black_box(unicode::lpad().invoke_with_args(ScalarFunctionArgs {
+                        args: args_cloned,
+                        arg_fields: arg_fields.clone(),
+                        number_rows: size,
+                        return_field: Field::new("f", DataType::Utf8, true).into(),
+                        config_options: Arc::clone(&config_options),
+                    }))
+                })
+            },
+        );
+
+        // Utf8 with scalar truncation (str_len > target) and unicode fill
+        let args = create_scalar_pad_args::<i32>(size, 20, 5, "é", false);
+        let arg_fields = args
+            .iter()
+            .enumerate()
+            .map(|(idx, arg)| {
+                Field::new(format!("arg_{idx}"), arg.data_type(), true).into()
+            })
+            .collect::<Vec<_>>();
+
+        group.bench_function(
+            format!(
+                "lpad utf8 scalar truncate [size={size}, str_len=20, target=5, fill='é']"
+            ),
+            |b| {
+                b.iter(|| {
+                    let args_cloned = args.clone();
+                    black_box(unicode::lpad().invoke_with_args(ScalarFunctionArgs {
+                        args: args_cloned,
+                        arg_fields: arg_fields.clone(),
+                        number_rows: size,
+                        return_field: Field::new("f", DataType::Utf8, true).into(),
+                        config_options: Arc::clone(&config_options),
+                    }))
+                })
+            },
+        );
+
         group.finish();
     }
 
@@ -502,6 +641,118 @@ fn criterion_benchmark(c: &mut Criterion) {
             },
         );
 
+        // --- Scalar length + fill benchmarks ---
+
+        // Utf8 with scalar length and fill (3-arg)
+        let args = create_scalar_pad_args::<i32>(size, 5, 20, "x", false);
+        let arg_fields = args
+            .iter()
+            .enumerate()
+            .map(|(idx, arg)| {
+                Field::new(format!("arg_{idx}"), arg.data_type(), true).into()
+            })
+            .collect::<Vec<_>>();
+
+        group.bench_function(
+            format!("rpad utf8 scalar [size={size}, str_len=5, target=20, fill='x']"),
+            |b| {
+                b.iter(|| {
+                    let args_cloned = args.clone();
+                    black_box(unicode::rpad().invoke_with_args(ScalarFunctionArgs {
+                        args: args_cloned,
+                        arg_fields: arg_fields.clone(),
+                        number_rows: size,
+                        return_field: Field::new("f", DataType::Utf8, true).into(),
+                        config_options: Arc::clone(&config_options),
+                    }))
+                })
+            },
+        );
+
+        // StringView with scalar length and fill (3-arg)
+        let args = create_scalar_pad_args::<i32>(size, 5, 20, "x", true);
+        let arg_fields = args
+            .iter()
+            .enumerate()
+            .map(|(idx, arg)| {
+                Field::new(format!("arg_{idx}"), arg.data_type(), true).into()
+            })
+            .collect::<Vec<_>>();
+
+        group.bench_function(
+            format!(
+                "rpad stringview scalar [size={size}, str_len=5, target=20, fill='x']"
+            ),
+            |b| {
+                b.iter(|| {
+                    let args_cloned = args.clone();
+                    black_box(unicode::rpad().invoke_with_args(ScalarFunctionArgs {
+                        args: args_cloned,
+                        arg_fields: arg_fields.clone(),
+                        number_rows: size,
+                        return_field: Field::new("f", DataType::Utf8, true).into(),
+                        config_options: Arc::clone(&config_options),
+                    }))
+                })
+            },
+        );
+
+        // Utf8 with scalar length and unicode fill
+        let args = create_scalar_pad_args::<i32>(size, 5, 20, "é", false);
+        let arg_fields = args
+            .iter()
+            .enumerate()
+            .map(|(idx, arg)| {
+                Field::new(format!("arg_{idx}"), arg.data_type(), true).into()
+            })
+            .collect::<Vec<_>>();
+
+        group.bench_function(
+            format!(
+                "rpad utf8 scalar unicode [size={size}, str_len=5, target=20, fill='é']"
+            ),
+            |b| {
+                b.iter(|| {
+                    let args_cloned = args.clone();
+                    black_box(unicode::rpad().invoke_with_args(ScalarFunctionArgs {
+                        args: args_cloned,
+                        arg_fields: arg_fields.clone(),
+                        number_rows: size,
+                        return_field: Field::new("f", DataType::Utf8, true).into(),
+                        config_options: Arc::clone(&config_options),
+                    }))
+                })
+            },
+        );
+
+        // Utf8 with scalar truncation (str_len > target) and unicode fill
+        let args = create_scalar_pad_args::<i32>(size, 20, 5, "é", false);
+        let arg_fields = args
+            .iter()
+            .enumerate()
+            .map(|(idx, arg)| {
+                Field::new(format!("arg_{idx}"), arg.data_type(), true).into()
+            })
+            .collect::<Vec<_>>();
+
+        group.bench_function(
+            format!(
+                "rpad utf8 scalar truncate [size={size}, str_len=20, target=5, fill='é']"
+            ),
+            |b| {
+                b.iter(|| {
+                    let args_cloned = args.clone();
+                    black_box(unicode::rpad().invoke_with_args(ScalarFunctionArgs {
+                        args: args_cloned,
+                        arg_fields: arg_fields.clone(),
+                        number_rows: size,
+                        return_field: Field::new("f", DataType::Utf8, true).into(),
+                        config_options: Arc::clone(&config_options),
+                    }))
+                })
+            },
+        );
+
         group.finish();
     }
 }

datafusion/functions/src/unicode/common.rs

@@ -23,14 +23,32 @@ use arrow::array::{
 };
 use arrow::datatypes::DataType;
 use arrow_buffer::{NullBuffer, ScalarBuffer};
+use datafusion_common::ScalarValue;
 use datafusion_common::cast::{
     as_generic_string_array, as_int64_array, as_string_view_array,
 };
 use datafusion_common::exec_err;
+use datafusion_expr::ColumnarValue;
 use std::cmp::Ordering;
 use std::ops::Range;
 use std::sync::Arc;
 
+/// If `cv` is a non-null scalar string, return its value.
+pub(crate) fn try_as_scalar_str(cv: &ColumnarValue) -> Option<&str> {
+    match cv {
+        ColumnarValue::Scalar(s) => s.try_as_str().flatten(),
+        _ => None,
+    }
+}
+
+/// If `cv` is a non-null scalar Int64, return its value.
+pub(crate) fn try_as_scalar_i64(cv: &ColumnarValue) -> Option<i64> {
+    match cv {
+        ColumnarValue::Scalar(ScalarValue::Int64(v)) => *v,
+        _ => None,
+    }
+}
+
 /// A trait for `left` and `right` byte slicing operations
 pub(crate) trait LeftRightSlicer {
     fn slice(string: &str, n: i64) -> Range<usize>;