diff --git a/datafusion/spark/Cargo.toml b/datafusion/spark/Cargo.toml index 93987b553f2f5..4a35323bf5591 100644 --- a/datafusion/spark/Cargo.toml +++ b/datafusion/spark/Cargo.toml @@ -102,6 +102,26 @@ name = "unhex" harness = false name = "sha2" +[[bench]] +harness = false +name = "bin" + +[[bench]] +harness = false +name = "soundex_quote" + +[[bench]] +harness = false +name = "url_encode" + +[[bench]] +harness = false +name = "next_day" + +[[bench]] +harness = false +name = "url_decode" + [[bench]] harness = false name = "floor" diff --git a/datafusion/spark/benches/bin.rs b/datafusion/spark/benches/bin.rs new file mode 100644 index 0000000000000..e65b2d74630ca --- /dev/null +++ b/datafusion/spark/benches/bin.rs @@ -0,0 +1,93 @@ +// 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 arrow::array::{ArrayRef, Int64Array}; +use arrow::datatypes::{DataType, Field}; +use criterion::{Criterion, criterion_group, criterion_main}; +use datafusion_common::config::ConfigOptions; +use datafusion_expr::{ColumnarValue, ScalarFunctionArgs}; +use datafusion_spark::function::math::bin; +use rand::rngs::StdRng; +use rand::{Rng, SeedableRng}; +use std::hint::black_box; +use std::sync::Arc; + +const NULL_DENSITY: f32 = 0.2; + +/// `range` controls how many binary digits each value renders to, which is what +/// drives the amount of work per row. +fn random_ints(size: usize, seed: u64, range: std::ops::Range) -> ArrayRef { + let mut rng = StdRng::seed_from_u64(seed); + let array: Int64Array = (0..size) + .map(|_| { + if rng.random::() < NULL_DENSITY { + None + } else { + Some(rng.random_range(range.clone())) + } + }) + .collect(); + Arc::new(array) +} + +fn bench(c: &mut Criterion, name: &str, input: ArrayRef) { + let bin_fn = bin(); + let size = input.len(); + let args = vec![ColumnarValue::Array(input)]; + let arg_fields = args + .iter() + .enumerate() + .map(|(idx, arg)| Field::new(format!("arg_{idx}"), arg.data_type(), true).into()) + .collect::>(); + let config_options = Arc::new(ConfigOptions::default()); + + c.bench_function(name, |b| { + b.iter(|| { + black_box( + bin_fn + .invoke_with_args(ScalarFunctionArgs { + args: args.clone(), + arg_fields: arg_fields.clone(), + number_rows: size, + return_field: Arc::new(Field::new("f", DataType::Utf8, true)), + config_options: Arc::clone(&config_options), + }) + .unwrap(), + ) + }) + }); +} + +fn criterion_benchmark(c: &mut Criterion) { + for size in [1024, 8192] { + // Small values render to a handful of digits; the common case. + bench( + c, + &format!("bin/small/size={size}"), + random_ints(size, 42, 0..10_000), + ); + // Full-width values render to the maximum 64 digits. + bench( + c, + &format!("bin/wide/size={size}"), + random_ints(size, 7, i64::MIN..i64::MAX), + ); + } +} + +criterion_group!(benches, criterion_benchmark); +criterion_main!(benches); diff --git a/datafusion/spark/benches/next_day.rs b/datafusion/spark/benches/next_day.rs new file mode 100644 index 0000000000000..0f0c653710921 --- /dev/null +++ b/datafusion/spark/benches/next_day.rs @@ -0,0 +1,119 @@ +// 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 arrow::array::{Date32Array, StringArray}; +use arrow::datatypes::{DataType, Field}; +use criterion::{Criterion, criterion_group, criterion_main}; +use datafusion_common::ScalarValue; +use datafusion_common::config::ConfigOptions; +use datafusion_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl}; +use datafusion_spark::function::datetime::next_day::SparkNextDay; +use rand::rngs::StdRng; +use rand::{Rng, SeedableRng}; +use std::hint::black_box; +use std::sync::Arc; + +const NULL_DENSITY: f32 = 0.2; + +/// Day names in mixed case, so the case-insensitive parse is actually exercised. +const DAY_NAMES: &[&str] = &[ + "MO", + "mon", + "Monday", + "TU", + "tue", + "WEDNESDAY", + "Th", + "fri", + "SAT", + "su", +]; + +fn random_dates(size: usize) -> Date32Array { + let mut rng = StdRng::seed_from_u64(42); + (0..size) + .map(|_| { + if rng.random::() < NULL_DENSITY { + None + } else { + // Roughly 1970-2040. + Some(rng.random_range(0i32..25_000)) + } + }) + .collect() +} + +fn bench(c: &mut Criterion, name: &str, args: Vec, size: usize) { + let func = SparkNextDay::new(); + let arg_fields = args + .iter() + .enumerate() + .map(|(idx, arg)| Field::new(format!("arg_{idx}"), arg.data_type(), true).into()) + .collect::>(); + let config_options = Arc::new(ConfigOptions::default()); + + c.bench_function(name, |b| { + b.iter(|| { + black_box( + func.invoke_with_args(ScalarFunctionArgs { + args: args.clone(), + arg_fields: arg_fields.clone(), + number_rows: size, + return_field: Arc::new(Field::new("f", DataType::Date32, true)), + config_options: Arc::clone(&config_options), + }) + .unwrap(), + ) + }) + }); +} + +fn criterion_benchmark(c: &mut Criterion) { + for size in [1024, 8192] { + let dates: Date32Array = random_dates(size); + + // The common form: `next_day(col, 'MONDAY')`. The day name is constant, + // so it need only be parsed once per batch. + bench( + c, + &format!("next_day/scalar_day/size={size}"), + vec![ + ColumnarValue::Array(Arc::new(dates.clone())), + ColumnarValue::Scalar(ScalarValue::Utf8(Some("MONDAY".to_string()))), + ], + size, + ); + + // Both arguments as columns, so the day name is parsed per row. + let mut rng = StdRng::seed_from_u64(7); + let days: StringArray = (0..size) + .map(|_| Some(DAY_NAMES[rng.random_range(0..DAY_NAMES.len())])) + .collect(); + bench( + c, + &format!("next_day/array_day/size={size}"), + vec![ + ColumnarValue::Array(Arc::new(dates.clone())), + ColumnarValue::Array(Arc::new(days)), + ], + size, + ); + } +} + +criterion_group!(benches, criterion_benchmark); +criterion_main!(benches); diff --git a/datafusion/spark/benches/soundex_quote.rs b/datafusion/spark/benches/soundex_quote.rs new file mode 100644 index 0000000000000..8a14a62093b95 --- /dev/null +++ b/datafusion/spark/benches/soundex_quote.rs @@ -0,0 +1,144 @@ +// 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 arrow::array::{ArrayRef, StringArray, StringViewArray}; +use arrow::datatypes::{DataType, Field}; +use criterion::{Criterion, criterion_group, criterion_main}; +use datafusion_common::config::ConfigOptions; +use datafusion_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDF}; +use datafusion_spark::function::string::{quote, soundex}; +use rand::rngs::StdRng; +use rand::{Rng, SeedableRng}; +use std::hint::black_box; +use std::sync::Arc; + +const NULL_DENSITY: f32 = 0.2; + +/// Words of varying length, so soundex sees both codes it truncates and codes it pads. +fn random_words(size: usize) -> Vec> { + let mut rng = StdRng::seed_from_u64(42); + (0..size) + .map(|_| { + if rng.random::() < NULL_DENSITY { + return None; + } + let len = rng.random_range(3..12); + Some( + (0..len) + .map(|_| rng.random_range(b'a'..=b'z') as char) + .collect(), + ) + }) + .collect() +} + +/// Sentences that mostly contain no quote at all, plus a minority that do, so the +/// escaping path is exercised without dominating. +fn random_quotable(size: usize) -> Vec> { + let mut rng = StdRng::seed_from_u64(7); + (0..size) + .map(|_| { + if rng.random::() < NULL_DENSITY { + return None; + } + let len = rng.random_range(8..40); + let has_quote = rng.random::() < 0.25; + Some( + (0..len) + .map(|i| { + if has_quote && i % 11 == 0 { + '\'' + } else { + rng.random_range(b'a'..=b'z') as char + } + }) + .collect(), + ) + }) + .collect() +} + +fn bench(c: &mut Criterion, name: &str, func: Arc, input: ArrayRef) { + let size = input.len(); + let args = vec![ColumnarValue::Array(input)]; + let arg_fields = args + .iter() + .enumerate() + .map(|(idx, arg)| Field::new(format!("arg_{idx}"), arg.data_type(), true).into()) + .collect::>(); + let config_options = Arc::new(ConfigOptions::default()); + + c.bench_function(name, |b| { + b.iter(|| { + black_box( + func.invoke_with_args(ScalarFunctionArgs { + args: args.clone(), + arg_fields: arg_fields.clone(), + number_rows: size, + return_field: Arc::new(Field::new("f", DataType::Utf8, true)), + config_options: Arc::clone(&config_options), + }) + .unwrap(), + ) + }) + }); +} + +fn criterion_benchmark(c: &mut Criterion) { + for size in [1024, 8192] { + let words = random_words(size); + bench( + c, + &format!("soundex/utf8/size={size}"), + soundex(), + Arc::new(words.iter().cloned().collect::()), + ); + bench( + c, + &format!("soundex/utf8view/size={size}"), + soundex(), + Arc::new( + words + .iter() + .map(|s| s.as_deref()) + .collect::(), + ), + ); + + let quotable = random_quotable(size); + bench( + c, + &format!("quote/utf8/size={size}"), + quote(), + Arc::new(quotable.iter().cloned().collect::()), + ); + bench( + c, + &format!("quote/utf8view/size={size}"), + quote(), + Arc::new( + quotable + .iter() + .map(|s| s.as_deref()) + .collect::(), + ), + ); + } +} + +criterion_group!(benches, criterion_benchmark); +criterion_main!(benches); diff --git a/datafusion/spark/benches/url_decode.rs b/datafusion/spark/benches/url_decode.rs new file mode 100644 index 0000000000000..0cf52bc25316c --- /dev/null +++ b/datafusion/spark/benches/url_decode.rs @@ -0,0 +1,125 @@ +// 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 arrow::array::{ArrayRef, LargeStringArray, StringArray, StringViewArray}; +use arrow::datatypes::Field; +use criterion::{Criterion, criterion_group, criterion_main}; +use datafusion_common::config::ConfigOptions; +use datafusion_expr::{ColumnarValue, ScalarFunctionArgs}; +use datafusion_spark::function::url::url_decode; +use rand::rngs::StdRng; +use rand::{Rng, SeedableRng}; +use std::hint::black_box; +use std::sync::Arc; + +const NULL_DENSITY: f32 = 0.2; + +/// Percent-escapes that appear in real encoded input. +const ESCAPES: &[&str] = &["%20", "%2F", "%3A", "%3F", "%26", "%3D", "%7E"]; + +/// Builds encoded strings where `escape_ratio` of the segments are percent-escapes. +/// +/// `escape_ratio == 0.0` produces input that needs no decoding at all, which is +/// the case where the decoded value can borrow rather than allocate. +fn random_encoded(size: usize, seed: u64, escape_ratio: f32) -> Vec> { + let mut rng = StdRng::seed_from_u64(seed); + (0..size) + .map(|_| { + if rng.random::() < NULL_DENSITY { + return None; + } + let segments = rng.random_range(4..12); + let mut s = String::new(); + for _ in 0..segments { + if rng.random::() < escape_ratio { + s.push_str(ESCAPES[rng.random_range(0..ESCAPES.len())]); + } else { + for _ in 0..rng.random_range(2..8) { + s.push(rng.random_range(b'a'..=b'z') as char); + } + } + } + Some(s) + }) + .collect() +} + +fn bench(c: &mut Criterion, name: &str, input: ArrayRef) { + let func = url_decode(); + let size = input.len(); + let return_type = input.data_type().clone(); + let args = vec![ColumnarValue::Array(input)]; + let arg_fields = args + .iter() + .enumerate() + .map(|(idx, arg)| Field::new(format!("arg_{idx}"), arg.data_type(), true).into()) + .collect::>(); + let config_options = Arc::new(ConfigOptions::default()); + + c.bench_function(name, |b| { + b.iter(|| { + black_box( + func.invoke_with_args(ScalarFunctionArgs { + args: args.clone(), + arg_fields: arg_fields.clone(), + number_rows: size, + return_field: Arc::new(Field::new("f", return_type.clone(), true)), + config_options: Arc::clone(&config_options), + }) + .unwrap(), + ) + }) + }); +} + +fn criterion_benchmark(c: &mut Criterion) { + for size in [1024, 8192] { + // Nothing to unescape: the decoded value can borrow its input. + let plain = random_encoded(size, 42, 0.0); + bench( + c, + &format!("url_decode/plain_utf8/size={size}"), + Arc::new(plain.iter().cloned().collect::()), + ); + bench( + c, + &format!("url_decode/plain_utf8view/size={size}"), + Arc::new( + plain + .iter() + .map(|s| s.as_deref()) + .collect::(), + ), + ); + + // A realistic mix, where roughly a third of segments need unescaping. + let escaped = random_encoded(size, 7, 0.33); + bench( + c, + &format!("url_decode/escaped_utf8/size={size}"), + Arc::new(escaped.iter().cloned().collect::()), + ); + bench( + c, + &format!("url_decode/escaped_largeutf8/size={size}"), + Arc::new(escaped.iter().cloned().collect::()), + ); + } +} + +criterion_group!(benches, criterion_benchmark); +criterion_main!(benches); diff --git a/datafusion/spark/benches/url_encode.rs b/datafusion/spark/benches/url_encode.rs new file mode 100644 index 0000000000000..ec3826eb5531a --- /dev/null +++ b/datafusion/spark/benches/url_encode.rs @@ -0,0 +1,106 @@ +// 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 arrow::array::{ArrayRef, LargeStringArray, StringArray, StringViewArray}; +use arrow::datatypes::Field; +use criterion::{Criterion, criterion_group, criterion_main}; +use datafusion_common::config::ConfigOptions; +use datafusion_expr::{ColumnarValue, ScalarFunctionArgs}; +use datafusion_spark::function::url::url_encode; +use rand::rngs::StdRng; +use rand::{Rng, SeedableRng}; +use std::hint::black_box; +use std::sync::Arc; + +const NULL_DENSITY: f32 = 0.2; + +/// Characters drawn from a set where most are passed through unencoded and a +/// minority need percent-escaping, which is the realistic mix for URL input. +const ALPHABET: &[u8] = b"abcdefghijklmnopqrstuvwxyz0123456789-_.~ &=/?#%+"; + +fn random_urls(size: usize) -> Vec> { + let mut rng = StdRng::seed_from_u64(42); + (0..size) + .map(|_| { + if rng.random::() < NULL_DENSITY { + return None; + } + let len = rng.random_range(10..60); + Some( + (0..len) + .map(|_| ALPHABET[rng.random_range(0..ALPHABET.len())] as char) + .collect(), + ) + }) + .collect() +} + +fn bench(c: &mut Criterion, name: &str, input: ArrayRef) { + let func = url_encode(); + let size = input.len(); + let return_type = input.data_type().clone(); + let args = vec![ColumnarValue::Array(input)]; + let arg_fields = args + .iter() + .enumerate() + .map(|(idx, arg)| Field::new(format!("arg_{idx}"), arg.data_type(), true).into()) + .collect::>(); + let config_options = Arc::new(ConfigOptions::default()); + + c.bench_function(name, |b| { + b.iter(|| { + black_box( + func.invoke_with_args(ScalarFunctionArgs { + args: args.clone(), + arg_fields: arg_fields.clone(), + number_rows: size, + return_field: Arc::new(Field::new("f", return_type.clone(), true)), + config_options: Arc::clone(&config_options), + }) + .unwrap(), + ) + }) + }); +} + +fn criterion_benchmark(c: &mut Criterion) { + for size in [1024, 8192] { + let urls = random_urls(size); + bench( + c, + &format!("url_encode/utf8/size={size}"), + Arc::new(urls.iter().cloned().collect::()), + ); + bench( + c, + &format!("url_encode/largeutf8/size={size}"), + Arc::new(urls.iter().cloned().collect::()), + ); + bench( + c, + &format!("url_encode/utf8view/size={size}"), + Arc::new( + urls.iter() + .map(|s| s.as_deref()) + .collect::(), + ), + ); + } +} + +criterion_group!(benches, criterion_benchmark); +criterion_main!(benches);