feat: scalar regex match physical expr

datafusion/physical-expr/Cargo.toml

@@ -53,6 +53,7 @@ itertools = { workspace = true, features = ["use_std"] }
 log = { workspace = true }
 paste = "^1.0"
 petgraph = "0.6.2"
+regex = { workspace = true }
 
 [dev-dependencies]
 arrow = { workspace = true, features = ["test_utils"] }
@@ -72,3 +73,7 @@ name = "case_when"
 [[bench]]
 harness = false
 name = "is_null"
+
+[[bench]]
+harness = false
+name = "scalar_regex_match"

datafusion/physical-expr/benches/scalar_regex_match.rs

@@ -0,0 +1,175 @@
+// 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 std::sync::Arc;
+use std::time::Duration;
+
+use arrow_array::{RecordBatch, StringArray};
+use arrow_schema::{DataType, Field, Schema};
+use criterion::{criterion_group, criterion_main, Criterion};
+use datafusion_expr_common::operator::Operator;
+use datafusion_physical_expr::{
+    expressions::{binary, col, lit, scalar_regex_match},
+    PhysicalExpr,
+};
+use rand::{
+    distributions::{Alphanumeric, DistString},
+    rngs::StdRng,
+    SeedableRng,
+};
+
+/// make a record batch with one column and n rows
+/// this record batch is single string column is used for
+/// scalar regex match benchmarks
+fn make_record_batch(
+    batch_iter: usize,
+    batch_size: usize,
+    string_len: usize,
+    matched_str: &[&str],
+    schema: &Schema,
+) -> Vec<RecordBatch> {
+    let mut rng = StdRng::seed_from_u64(12345);
+    let mut batches = vec![];
+    for _ in 0..batch_iter {
+        let mut array = (0..batch_size)
+            .map(|_| Some(Alphanumeric.sample_string(&mut rng, string_len)))
+            .collect::<Vec<_>>();
+        for v in matched_str {
+            array.push(Some(v.to_string()));
+        }
+        let array = StringArray::from(array);
+        let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(array)])
+            .unwrap();
+        batches.push(batch);
+    }
+    batches
+}
+
+/// initialize benchmark data and pattern literals
+#[allow(clippy::type_complexity)]
+fn init_benchmark() -> (
+    Vec<(usize, usize, Vec<RecordBatch>)>,
+    Schema,
+    Arc<dyn PhysicalExpr>,
+    Vec<(String, Arc<dyn PhysicalExpr>)>,
+) {
+    // make common schema
+    let column = "s";
+    let schema = Schema::new(vec![Field::new(column, DataType::Utf8, true)]);
+
+    // make test record batch
+    let batch_data = vec![
+        // (20, 10_usize, make_record_batch(20, 10, 100, schema.clone())),
+        // (20, 100_usize, make_record_batch(20, 100, 100, schema.clone())),
+        // (20, 1000_usize, make_record_batch(20, 1000, 100, schema.clone())),
+        (
+            128_usize,
+            4096_usize,
+            make_record_batch(
+                128,
+                4096,
+                100,
+                &[
+                    "example@email.com",
+                    "http://example.com",
+                    "123.4.5.6",
+                    "1236787788",
+                    "55555",
+                ],
+                &schema,
+            ),
+        ),
+    ];
+
+    // string column
+    let string_col = col(column, &schema).unwrap();
+
+    // some pattern literal
+    let pattern_lit = vec![
+        (
+            "email".to_string(),
+            lit(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"),
+        ),
+        (
+            "url".to_string(),
+            lit(r"^(https?|ftp)://[-a-zA-Z0-9+&@#/%?=~_|!:,.;]*[-a-zA-Z0-9+&@#/%=~_|]$"),
+        ),
+        (
+            "ip".to_string(),
+            lit(
+                r"^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$",
+            ),
+        ),
+        (
+            "phone".to_string(),
+            lit(r"^(\+\d{1,2}\s?)?\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4}$"),
+        ),
+        ("zip_code".to_string(), lit(r"^\d{5}(?:[-\s]\d{4})?$")),
+    ];
+    (batch_data, schema, string_col, pattern_lit)
+}
+
+fn regex_match_benchmark(c: &mut Criterion) {
+    let (batch_data, schema, string_col, pattern_lit) = init_benchmark();
+    for (name, regexp_lit) in pattern_lit.iter() {
+        for (batch_iter, batch_size, batches) in batch_data.iter() {
+            let group_name = format!(
+                "regex_{}_batch_iter_{}_batch_size_{}",
+                name, batch_iter, batch_size
+            );
+            let mut group = c.benchmark_group(group_name.as_str());
+
+            group.sample_size(50).measurement_time(Duration::new(30, 0));
+
+            // binary expr match benchmarks
+            group.bench_function("binary_expr_match", |b| {
+                b.iter(|| {
+                    let expr = binary(
+                        string_col.clone(),
+                        Operator::RegexMatch,
+                        regexp_lit.clone(),
+                        &schema,
+                    )
+                    .unwrap();
+                    for batch in batches.iter() {
+                        expr.evaluate(batch).unwrap();
+                    }
+                });
+            });
+            // scalar regex match benchmarks
+            group.bench_function("scalar_regex_match", |b| {
+                b.iter(|| {
+                    let expr = scalar_regex_match(
+                        false,
+                        false,
+                        string_col.clone(),
+                        regexp_lit.clone(),
+                        &schema,
+                    )
+                    .unwrap();
+                    for batch in batches.iter() {
+                        expr.evaluate(batch).unwrap();
+                    }
+                });
+            });
+            group.finish();
+        }
+    }
+}
+
+criterion_group!(benches, regex_match_benchmark);
+criterion_main!(benches);

datafusion/physical-expr/src/expressions/mod.rs

@@ -30,6 +30,7 @@ mod literal;
 mod negative;
 mod no_op;
 mod not;
+mod scalar_regex_match;
 mod try_cast;
 mod unknown_column;
 
@@ -50,5 +51,6 @@ pub use literal::{lit, Literal};
 pub use negative::{negative, NegativeExpr};
 pub use no_op::NoOp;
 pub use not::{not, NotExpr};
+pub use scalar_regex_match::{scalar_regex_match, ScalarRegexMatchExpr};
 pub use try_cast::{try_cast, TryCastExpr};
 pub use unknown_column::UnKnownColumn;