From 24e5a4e2516c9b3c5337bcfbc78745b20ec23fe3 Mon Sep 17 00:00:00 2001 From: sam1112 Date: Sun, 2 Aug 2026 21:33:46 +0800 Subject: [PATCH 1/4] fix: support Utf8/LargeUtf8/Utf8View in native RLike without panicking (#5102) --- .../spark-expr/src/predicate_funcs/rlike.rs | 125 +++++++++++++++--- 1 file changed, 104 insertions(+), 21 deletions(-) diff --git a/native/spark-expr/src/predicate_funcs/rlike.rs b/native/spark-expr/src/predicate_funcs/rlike.rs index ee005dd1ac..5a10ef13fa 100644 --- a/native/spark-expr/src/predicate_funcs/rlike.rs +++ b/native/spark-expr/src/predicate_funcs/rlike.rs @@ -18,10 +18,15 @@ use crate::SparkError; use arrow::array::builder::BooleanBuilder; use arrow::array::types::Int32Type; -use arrow::array::{Array, BooleanArray, DictionaryArray, RecordBatch, StringArray}; +use arrow::array::{ + Array, ArrayAccessor, ArrayRef, BooleanArray, RecordBatch, StringArrayType, +}; use arrow::compute::take; use arrow::datatypes::{DataType, Schema}; -use datafusion::common::{internal_err, Result, ScalarValue}; +use datafusion::common::cast::{ + as_dictionary_array, as_large_string_array, as_string_array, as_string_view_array, +}; +use datafusion::common::{exec_err, internal_err, Result, ScalarValue}; use datafusion::physical_expr::PhysicalExpr; use datafusion::physical_plan::ColumnarValue; use regex::Regex; @@ -37,6 +42,11 @@ use std::sync::Arc; /// regular expression engine, which are documented at: /// /// https://docs.oracle.com/javase/8/docs/api/java/util/regex/Pattern.html +/// +/// Array matching keeps the plan-time compiled [`Regex`] and loops over Utf8 / +/// LargeUtf8 / Utf8View inputs. Arrow's `regexp_is_match(_scalar)` was considered +/// but recompiles the pattern per batch; criterion benches showed regressions on +/// common patterns such as character classes (see issue #5102). #[derive(Debug)] pub struct RLike { child: Arc, @@ -71,7 +81,14 @@ impl RLike { }) } - fn is_match(&self, inputs: &StringArray) -> BooleanArray { + /// Match the pre-compiled pattern against a string array of any Arrow string layout. + /// + /// Keeps the plan-time compiled [`Regex`] rather than calling Arrow's + /// `regexp_is_match(_scalar)`, which recompiles the pattern on every batch. + fn is_match<'a, S>(&'a self, inputs: &'a S) -> BooleanArray + where + &'a S: StringArrayType<'a>, + { let mut builder = BooleanBuilder::with_capacity(inputs.len()); if inputs.is_nullable() { for i in 0..inputs.len() { @@ -88,6 +105,15 @@ impl RLike { } builder.finish() } + + fn is_match_array(&self, array: &ArrayRef) -> Result { + match array.data_type() { + DataType::Utf8 => Ok(self.is_match(as_string_array(array)?)), + DataType::LargeUtf8 => Ok(self.is_match(as_large_string_array(array)?)), + DataType::Utf8View => Ok(self.is_match(as_string_view_array(array)?)), + other => exec_err!("RLike requires string type for input, got {other:?}"), + } + } } impl Display for RLike { @@ -111,29 +137,19 @@ impl PhysicalExpr for RLike { fn evaluate(&self, batch: &RecordBatch) -> Result { match self.child.evaluate(batch)? { - ColumnarValue::Array(array) if array.as_any().is::>() => { - let dict_array = array - .as_any() - .downcast_ref::>() - .expect("dict array"); - let dict_values = dict_array - .values() - .as_any() - .downcast_ref::() - .expect("strings"); + ColumnarValue::Array(array) + if matches!(array.data_type(), DataType::Dictionary(_, _)) => + { + let dict_array = as_dictionary_array::(&array)?; // evaluate the regexp pattern against the dictionary values - let new_values = self.is_match(dict_values); + let new_values = self.is_match_array(dict_array.values())?; // convert to conventional (not dictionary-encoded) array let result = take(&new_values, dict_array.keys(), None)?; Ok(ColumnarValue::Array(result)) } ColumnarValue::Array(array) => { - let inputs = array - .as_any() - .downcast_ref::() - .expect("string array"); - let array = self.is_match(inputs); - Ok(ColumnarValue::Array(Arc::new(array))) + let result = self.is_match_array(&array)?; + Ok(ColumnarValue::Array(Arc::new(result))) } ColumnarValue::Scalar(scalar) => { if scalar.is_null() { @@ -180,7 +196,22 @@ impl PhysicalExpr for RLike { #[cfg(test)] mod tests { use super::*; - use datafusion::physical_expr::expressions::Literal; + use arrow::array::{LargeStringArray, StringArray, StringViewArray}; + use arrow::datatypes::Field; + use datafusion::physical_expr::expressions::{Column, Literal}; + + fn assert_bool_array_rose_null(result: ColumnarValue) { + let ColumnarValue::Array(arr) = result else { + panic!("expected array result"); + }; + let bools = arr + .as_any() + .downcast_ref::() + .expect("boolean array"); + assert_eq!(bools.len(), 2); + assert!(bools.value(0)); + assert!(bools.is_null(1)); + } #[test] fn test_rlike_scalar_string_variants() { @@ -225,4 +256,56 @@ mod tests { let result = expr.evaluate(&RecordBatch::new_empty(Arc::new(Schema::empty()))); assert!(result.is_err()); } + + #[test] + fn test_rlike_utf8_array() { + let schema = Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8, true)])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(StringArray::from(vec![Some("Rose"), None]))], + ) + .unwrap(); + + let expr = RLike::try_new(Arc::new(Column::new("s", 0)), "R[a-z]+").unwrap(); + assert_bool_array_rose_null(expr.evaluate(&batch).unwrap()); + } + + #[test] + fn test_rlike_large_utf8_array() { + let schema = Arc::new(Schema::new(vec![Field::new("s", DataType::LargeUtf8, true)])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(LargeStringArray::from(vec![Some("Rose"), None]))], + ) + .unwrap(); + + let expr = RLike::try_new(Arc::new(Column::new("s", 0)), "R[a-z]+").unwrap(); + assert_bool_array_rose_null(expr.evaluate(&batch).unwrap()); + } + + #[test] + fn test_rlike_utf8_view_array() { + let schema = Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8View, true)])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(StringViewArray::from(vec![Some("Rose"), None]))], + ) + .unwrap(); + + let expr = RLike::try_new(Arc::new(Column::new("s", 0)), "R[a-z]+").unwrap(); + assert_bool_array_rose_null(expr.evaluate(&batch).unwrap()); + } + + #[test] + fn test_rlike_array_non_string_error() { + let schema = Arc::new(Schema::new(vec![Field::new("b", DataType::Boolean, true)])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(BooleanArray::from(vec![Some(true), None]))], + ) + .unwrap(); + + let expr = RLike::try_new(Arc::new(Column::new("b", 0)), "R[a-z]+").unwrap(); + assert!(expr.evaluate(&batch).is_err()); + } } From 00f6ca0b4b9b5464a3071f5256235c2c8431d3f6 Mon Sep 17 00:00:00 2001 From: sam1112 Date: Sun, 2 Aug 2026 22:01:51 +0800 Subject: [PATCH 2/4] style: cargo fmt --- native/spark-expr/src/predicate_funcs/rlike.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/native/spark-expr/src/predicate_funcs/rlike.rs b/native/spark-expr/src/predicate_funcs/rlike.rs index 5a10ef13fa..4a3cb457ab 100644 --- a/native/spark-expr/src/predicate_funcs/rlike.rs +++ b/native/spark-expr/src/predicate_funcs/rlike.rs @@ -18,9 +18,7 @@ use crate::SparkError; use arrow::array::builder::BooleanBuilder; use arrow::array::types::Int32Type; -use arrow::array::{ - Array, ArrayAccessor, ArrayRef, BooleanArray, RecordBatch, StringArrayType, -}; +use arrow::array::{Array, ArrayAccessor, ArrayRef, BooleanArray, RecordBatch, StringArrayType}; use arrow::compute::take; use arrow::datatypes::{DataType, Schema}; use datafusion::common::cast::{ @@ -272,7 +270,11 @@ mod tests { #[test] fn test_rlike_large_utf8_array() { - let schema = Arc::new(Schema::new(vec![Field::new("s", DataType::LargeUtf8, true)])); + let schema = Arc::new(Schema::new(vec![Field::new( + "s", + DataType::LargeUtf8, + true, + )])); let batch = RecordBatch::try_new( Arc::clone(&schema), vec![Arc::new(LargeStringArray::from(vec![Some("Rose"), None]))], From 4874f489878448306e77dc7ce8f362e6f377f4a8 Mon Sep 17 00:00:00 2001 From: sam1112 Date: Tue, 4 Aug 2026 01:10:54 +0800 Subject: [PATCH 3/4] fix: handle any dictionary key type in RLike and tighten review follow-ups Use as_any_dictionary() so non-Int32 keys work, unify non-string errors on dictionary layouts and string array coverage. --- .../spark-expr/src/predicate_funcs/rlike.rs | 161 +++++++++++------- 1 file changed, 100 insertions(+), 61 deletions(-) diff --git a/native/spark-expr/src/predicate_funcs/rlike.rs b/native/spark-expr/src/predicate_funcs/rlike.rs index 4a3cb457ab..d9ac380279 100644 --- a/native/spark-expr/src/predicate_funcs/rlike.rs +++ b/native/spark-expr/src/predicate_funcs/rlike.rs @@ -16,15 +16,11 @@ // under the License. use crate::SparkError; -use arrow::array::builder::BooleanBuilder; -use arrow::array::types::Int32Type; -use arrow::array::{Array, ArrayAccessor, ArrayRef, BooleanArray, RecordBatch, StringArrayType}; +use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, RecordBatch, StringArrayType}; use arrow::compute::take; use arrow::datatypes::{DataType, Schema}; -use datafusion::common::cast::{ - as_dictionary_array, as_large_string_array, as_string_array, as_string_view_array, -}; -use datafusion::common::{exec_err, internal_err, Result, ScalarValue}; +use datafusion::common::cast::{as_large_string_array, as_string_array, as_string_view_array}; +use datafusion::common::{internal_err, Result, ScalarValue}; use datafusion::physical_expr::PhysicalExpr; use datafusion::physical_plan::ColumnarValue; use regex::Regex; @@ -41,10 +37,6 @@ use std::sync::Arc; /// /// https://docs.oracle.com/javase/8/docs/api/java/util/regex/Pattern.html /// -/// Array matching keeps the plan-time compiled [`Regex`] and loops over Utf8 / -/// LargeUtf8 / Utf8View inputs. Arrow's `regexp_is_match(_scalar)` was considered -/// but recompiles the pattern per batch; criterion benches showed regressions on -/// common patterns such as character classes (see issue #5102). #[derive(Debug)] pub struct RLike { child: Arc, @@ -83,25 +75,14 @@ impl RLike { /// /// Keeps the plan-time compiled [`Regex`] rather than calling Arrow's /// `regexp_is_match(_scalar)`, which recompiles the pattern on every batch. - fn is_match<'a, S>(&'a self, inputs: &'a S) -> BooleanArray + fn is_match<'a, S>(&self, inputs: &'a S) -> BooleanArray where &'a S: StringArrayType<'a>, { - let mut builder = BooleanBuilder::with_capacity(inputs.len()); - if inputs.is_nullable() { - for i in 0..inputs.len() { - if inputs.is_null(i) { - builder.append_null(); - } else { - builder.append_value(self.pattern.is_match(inputs.value(i))); - } - } - } else { - for i in 0..inputs.len() { - builder.append_value(self.pattern.is_match(inputs.value(i))); - } - } - builder.finish() + inputs + .iter() + .map(|v| v.map(|s| self.pattern.is_match(s))) + .collect() } fn is_match_array(&self, array: &ArrayRef) -> Result { @@ -109,7 +90,9 @@ impl RLike { DataType::Utf8 => Ok(self.is_match(as_string_array(array)?)), DataType::LargeUtf8 => Ok(self.is_match(as_large_string_array(array)?)), DataType::Utf8View => Ok(self.is_match(as_string_view_array(array)?)), - other => exec_err!("RLike requires string type for input, got {other:?}"), + other => { + internal_err!("RLike requires string type for input, got {other:?}") + } } } } @@ -138,7 +121,7 @@ impl PhysicalExpr for RLike { ColumnarValue::Array(array) if matches!(array.data_type(), DataType::Dictionary(_, _)) => { - let dict_array = as_dictionary_array::(&array)?; + let dict_array = array.as_any_dictionary(); // evaluate the regexp pattern against the dictionary values let new_values = self.is_match_array(dict_array.values())?; // convert to conventional (not dictionary-encoded) array @@ -194,11 +177,13 @@ impl PhysicalExpr for RLike { #[cfg(test)] mod tests { use super::*; - use arrow::array::{LargeStringArray, StringArray, StringViewArray}; - use arrow::datatypes::Field; + use arrow::array::{ + DictionaryArray, Int32Array, Int8Array, LargeStringArray, StringArray, StringViewArray, + }; + use arrow::datatypes::{Field, Int32Type, Int8Type}; use datafusion::physical_expr::expressions::{Column, Literal}; - fn assert_bool_array_rose_null(result: ColumnarValue) { + fn assert_bool_results(result: ColumnarValue, expected: &[Option]) { let ColumnarValue::Array(arr) = result else { panic!("expected array result"); }; @@ -206,9 +191,16 @@ mod tests { .as_any() .downcast_ref::() .expect("boolean array"); - assert_eq!(bools.len(), 2); - assert!(bools.value(0)); - assert!(bools.is_null(1)); + assert_eq!(bools.len(), expected.len()); + for (i, exp) in expected.iter().enumerate() { + match exp { + Some(v) => { + assert!(!bools.is_null(i), "row {i} should not be null"); + assert_eq!(bools.value(i), *v, "row {i}"); + } + None => assert!(bools.is_null(i), "row {i} should be null"), + } + } } #[test] @@ -256,46 +248,93 @@ mod tests { } #[test] - fn test_rlike_utf8_array() { - let schema = Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8, true)])); - let batch = RecordBatch::try_new( - Arc::clone(&schema), - vec![Arc::new(StringArray::from(vec![Some("Rose"), None]))], - ) - .unwrap(); + fn test_rlike_string_array_layouts() { + let pattern = "R[a-z]+"; + let cases: Vec<(DataType, ArrayRef)> = vec![ + ( + DataType::Utf8, + Arc::new(StringArray::from(vec![Some("Rose"), None, Some("Daisy")])), + ), + ( + DataType::LargeUtf8, + Arc::new(LargeStringArray::from(vec![ + Some("Rose"), + None, + Some("Daisy"), + ])), + ), + ( + DataType::Utf8View, + Arc::new(StringViewArray::from(vec![ + Some("Rose"), + None, + Some("Daisy"), + ])), + ), + ]; - let expr = RLike::try_new(Arc::new(Column::new("s", 0)), "R[a-z]+").unwrap(); - assert_bool_array_rose_null(expr.evaluate(&batch).unwrap()); + for (data_type, array) in cases { + let schema = Arc::new(Schema::new(vec![Field::new("s", data_type, true)])); + let batch = RecordBatch::try_new(Arc::clone(&schema), vec![array]).unwrap(); + let expr = RLike::try_new(Arc::new(Column::new("s", 0)), pattern).unwrap(); + assert_bool_results( + expr.evaluate(&batch).unwrap(), + &[Some(true), None, Some(false)], + ); + } } #[test] - fn test_rlike_large_utf8_array() { - let schema = Arc::new(Schema::new(vec![Field::new( - "s", - DataType::LargeUtf8, - true, - )])); + fn test_rlike_string_array_no_nulls() { + let schema = Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8, false)])); let batch = RecordBatch::try_new( Arc::clone(&schema), - vec![Arc::new(LargeStringArray::from(vec![Some("Rose"), None]))], + vec![Arc::new(StringArray::from(vec!["Rose", "Daisy"]))], ) .unwrap(); let expr = RLike::try_new(Arc::new(Column::new("s", 0)), "R[a-z]+").unwrap(); - assert_bool_array_rose_null(expr.evaluate(&batch).unwrap()); + assert_bool_results(expr.evaluate(&batch).unwrap(), &[Some(true), Some(false)]); } #[test] - fn test_rlike_utf8_view_array() { - let schema = Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8View, true)])); - let batch = RecordBatch::try_new( - Arc::clone(&schema), - vec![Arc::new(StringViewArray::from(vec![Some("Rose"), None]))], - ) - .unwrap(); + fn test_rlike_dictionary_arrays() { + let pattern = "R[a-z]+"; + let expected = [Some(true), None, Some(false)]; + + let utf8_values: ArrayRef = Arc::new(StringArray::from(vec!["Rose", "Daisy"])); + let utf8_view_values: ArrayRef = Arc::new(StringViewArray::from(vec!["Rose", "Daisy"])); + + let cases: Vec<(DataType, ArrayRef)> = vec![ + ( + DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)), + Arc::new(DictionaryArray::::new( + Int32Array::from(vec![Some(0), None, Some(1)]), + Arc::clone(&utf8_values), + )), + ), + ( + DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8View)), + Arc::new(DictionaryArray::::new( + Int32Array::from(vec![Some(0), None, Some(1)]), + Arc::clone(&utf8_view_values), + )), + ), + ( + DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)), + Arc::new(DictionaryArray::::new( + Int8Array::from(vec![Some(0), None, Some(1)]), + Arc::clone(&utf8_values), + )), + ), + ]; - let expr = RLike::try_new(Arc::new(Column::new("s", 0)), "R[a-z]+").unwrap(); - assert_bool_array_rose_null(expr.evaluate(&batch).unwrap()); + for (data_type, array) in cases { + let schema = Arc::new(Schema::new(vec![Field::new("s", data_type, true)])); + let batch = RecordBatch::try_new(Arc::clone(&schema), vec![array]).unwrap(); + let expr = RLike::try_new(Arc::new(Column::new("s", 0)), pattern).unwrap(); + assert_bool_results(expr.evaluate(&batch).unwrap(), &expected); + } } #[test] From c799e5bd8d707aefdf8983fbc83cd8f820f0eb12 Mon Sep 17 00:00:00 2001 From: sam1112 Date: Tue, 4 Aug 2026 16:39:15 +0800 Subject: [PATCH 4/4] test: cover nulls in RLike dictionary values and drop stray doc comment --- native/spark-expr/src/predicate_funcs/rlike.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/native/spark-expr/src/predicate_funcs/rlike.rs b/native/spark-expr/src/predicate_funcs/rlike.rs index d9ac380279..dbf25d3f7a 100644 --- a/native/spark-expr/src/predicate_funcs/rlike.rs +++ b/native/spark-expr/src/predicate_funcs/rlike.rs @@ -36,7 +36,6 @@ use std::sync::Arc; /// regular expression engine, which are documented at: /// /// https://docs.oracle.com/javase/8/docs/api/java/util/regex/Pattern.html -/// #[derive(Debug)] pub struct RLike { child: Arc, @@ -304,6 +303,9 @@ mod tests { let utf8_values: ArrayRef = Arc::new(StringArray::from(vec!["Rose", "Daisy"])); let utf8_view_values: ArrayRef = Arc::new(StringViewArray::from(vec!["Rose", "Daisy"])); + // Null in dictionary values (keys all valid): is_match emits null, take carries it. + let utf8_values_with_null: ArrayRef = + Arc::new(StringArray::from(vec![Some("Rose"), None, Some("Daisy")])); let cases: Vec<(DataType, ArrayRef)> = vec![ ( @@ -327,6 +329,13 @@ mod tests { Arc::clone(&utf8_values), )), ), + ( + DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)), + Arc::new(DictionaryArray::::new( + Int32Array::from(vec![Some(0), Some(1), Some(2)]), + utf8_values_with_null, + )), + ), ]; for (data_type, array) in cases {