This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new b9cbb6882b perf: optimize parse_url in datafusion-comet-spark-expr
(#4893)
b9cbb6882b is described below
commit b9cbb6882b6aa2f5c7e6a77dd63ddf471d44ece9
Author: Andy Grove <[email protected]>
AuthorDate: Sat Jul 11 11:20:21 2026 -0600
perf: optimize parse_url in datafusion-comet-spark-expr (#4893)
---
native/spark-expr/Cargo.toml | 4 ++
native/spark-expr/benches/parse_url.rs | 74 ++++++++++++++++++++++++++++
native/spark-expr/src/url_funcs/parse_url.rs | 32 +++++++++---
3 files changed, 103 insertions(+), 7 deletions(-)
diff --git a/native/spark-expr/Cargo.toml b/native/spark-expr/Cargo.toml
index 4addc026d8..ef00bef1ea 100644
--- a/native/spark-expr/Cargo.toml
+++ b/native/spark-expr/Cargo.toml
@@ -131,3 +131,7 @@ harness = false
[[bench]]
name = "array_size"
harness = false
+
+[[bench]]
+name = "parse_url"
+harness = false
diff --git a/native/spark-expr/benches/parse_url.rs
b/native/spark-expr/benches/parse_url.rs
new file mode 100644
index 0000000000..5be9c36f86
--- /dev/null
+++ b/native/spark-expr/benches/parse_url.rs
@@ -0,0 +1,74 @@
+// 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::StringArray;
+use arrow::datatypes::{DataType, Field};
+use criterion::{criterion_group, criterion_main, Criterion};
+use datafusion::common::config::ConfigOptions;
+use datafusion::logical_expr::{ColumnarValue, ScalarFunctionArgs,
ScalarUDFImpl};
+use datafusion_comet_spark_expr::url_funcs::CometParseUrl;
+use std::sync::Arc;
+
+const N: usize = 10_000;
+
+fn call(udf: &CometParseUrl, args: Vec<ColumnarValue>) -> ColumnarValue {
+ let return_field = Arc::new(Field::new("parse_url", DataType::Utf8, true));
+ let sfa = ScalarFunctionArgs {
+ args,
+ number_rows: N,
+ return_field,
+ config_options: Arc::new(ConfigOptions::default()),
+ arg_fields: vec![],
+ };
+ udf.invoke_with_args(sfa).unwrap()
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+ let mut group = c.benchmark_group("parse_url");
+
+ let udf = CometParseUrl::default();
+
+ // The common Spark shape: extract a query parameter using a constant key
+ // (`parse_url(col, 'QUERY', 'v')`). The key is broadcast to every row.
+ let urls = StringArray::from(
+ (0..N)
+ .map(|i|
format!("https://example.com/path?a={i}&v=value{i}&b={i}";))
+ .collect::<Vec<_>>(),
+ );
+ let urls = ColumnarValue::Array(Arc::new(urls));
+ let part =
ColumnarValue::Scalar(datafusion::common::ScalarValue::Utf8(Some(
+ "QUERY".to_string(),
+ )));
+ let key =
ColumnarValue::Scalar(datafusion::common::ScalarValue::Utf8(Some("v".to_string())));
+
+ group.bench_function("query_key", |b| {
+ b.iter(|| call(&udf, vec![urls.clone(), part.clone(), key.clone()]));
+ });
+
+ group.finish();
+}
+
+fn config() -> Criterion {
+ Criterion::default()
+}
+
+criterion_group! {
+ name = benches;
+ config = config();
+ targets = criterion_benchmark
+}
+criterion_main!(benches);
diff --git a/native/spark-expr/src/url_funcs/parse_url.rs
b/native/spark-expr/src/url_funcs/parse_url.rs
index 25d4ca94fe..7e072b6895 100644
--- a/native/spark-expr/src/url_funcs/parse_url.rs
+++ b/native/spark-expr/src/url_funcs/parse_url.rs
@@ -21,6 +21,7 @@
//! which diverges from Spark's java.net.URI (RFC 3986) on several edge cases.
//! This module uses RFC 3986 Appendix B regex parsing to match Spark exactly.
+use std::cell::RefCell;
use std::sync::{Arc, LazyLock};
use arrow::array::{
@@ -73,16 +74,33 @@ fn extract_userinfo(authority: &str) -> Option<String> {
authority.rfind('@').map(|pos| authority[..pos].to_string())
}
+thread_local! {
+ // The query regex is derived solely from `key`, which is almost always
+ // constant across a batch (e.g. `parse_url(col, 'QUERY', 'v')`). Naive
code
+ // recompiles it for every row; cache the last (key, compiled regex) pair
so
+ // it is compiled once per distinct key instead. A `None` regex records
that
+ // `key` failed to compile as a pattern.
+ static QUERY_KEY_REGEX: RefCell<Option<(String, Option<Regex>)>> =
+ const { RefCell::new(None) };
+}
+
fn extract_query_value(query: &str, key: &str) -> Option<String> {
// Spark uses Pattern.compile("(&|^)" + key + "=([^&]*)") with no escaping,
// so the key is treated as a regex pattern.
- let pattern = format!("(&|^){}=([^&]*)", key);
- match Regex::new(&pattern) {
- Ok(re) => re
- .captures(query)
- .and_then(|caps| caps.get(2).map(|m| m.as_str().to_string())),
- Err(_) => None,
- }
+ QUERY_KEY_REGEX.with(|cell| {
+ let mut slot = cell.borrow_mut();
+ // Drop the cached entry when the key changes, then (re)build lazily.
+ if slot.as_ref().map(|(k, _)| k.as_str()) != Some(key) {
+ *slot = None;
+ }
+ let (_, re) = slot.get_or_insert_with(|| {
+ let pattern = format!("(&|^){}=([^&]*)", key);
+ (key.to_string(), Regex::new(&pattern).ok())
+ });
+ let re = re.as_ref()?;
+ re.captures(query)
+ .and_then(|caps| caps.get(2).map(|m| m.as_str().to_string()))
+ })
}
fn has_invalid_uri_chars(s: &str) -> bool {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
