sunchao commented on code in PR #5416:
URL: https://github.com/apache/datafusion-comet/pull/5416#discussion_r4104427485
##########
native/spark-expr/src/string_funcs/split.rs:
##########
@@ -102,9 +106,47 @@ pub fn spark_split(
}
};
- let result = split_string(string.as_ref().unwrap(), pattern_str,
limit, regex_cache)?;
- let string_array = GenericStringArray::<i32>::from(result);
- let list_array = create_list_array(Arc::new(string_array));
+ let s = string.as_ref().unwrap();
+
+ let mut str_offsets = BufferBuilder::<i32>::new(8);
+ let mut str_values = BufferBuilder::<u8>::new(s.len());
+ str_offsets.append(0);
+
+ if is_regex_literal(pattern_str) {
+ let mut chars = pattern_str.chars();
+ if let (Some(ch), None) = (chars.next(), chars.next()) {
+ push_split_char(s, ch, limit, &mut str_offsets, &mut
str_values);
+ } else {
+ push_split_literal(s, pattern_str, limit, &mut
str_offsets, &mut str_values);
+ }
+ } else {
+ let regex = Regex::new(pattern_str).map_err(|e| {
Review Comment:
[P2] [P2] Preserve the existing cache in the scalar regex branch
Replacing the cached helper with `Regex::new` recompiles the same pattern on
every scalar evaluation. This affects native opt-in queries such as `SELECT
split((SELECT max(s) FROM t), r'\d+') FROM t`, where the subquery supplies a
scalar for each batch. The `LargeUtf8` array branch at line 406 also loses
caching. The base compiled these patterns once per expression. Please use the
supplied `regex_cache.get_or_compile` in both branches, as the `Utf8` array
branch already does.
Evidence: An exact-base/head module probe reused one `PatternCache`, warmed
up ten calls, then evaluated scalar `foo123bar456baz` with pattern `\d+` 1,000
times. Two samples measured base 191.9/190.8 ms versus head 920.4/934.9 ms. The
same regression reproduced with one-row `LargeUtf8` arrays. These are focused
measurements with debug dependencies, not release throughput estimates. Source
routing confirms that native string subqueries return
`ColumnarValue::Scalar(Utf8)`.
##########
native/spark-expr/src/string_funcs/split.rs:
##########
@@ -363,7 +601,12 @@ fn split_sql_generic_scalar_array<O: OffsetSizeTrait, D:
OffsetSizeTrait>(
) -> DataFusionResult<ColumnarValue> {
let len = delimiter_array.len();
let mut offsets: Vec<O> = Vec::with_capacity(len + 1);
- let mut values_builder = GenericStringBuilder::<O>::new();
+
+ let estimated_items = (len * 4).max(16);
+ let bytes_capacity = string.len() * len;
Review Comment:
[P2] [P2] Bound preallocation by data that contributes to the result
This reserves `string.len() * batch_rows` before checking delimiter nulls.
For `split_part(repeat('x', 8192), d, 1)` with an 8,192-row null delimiter
column, no string values are emitted, yet this intermediate result allocates
over 64 MiB. Previously it needed only about 38 KiB. The array branches
similarly reserve the whole backing buffer through `value_data().len()`,
including data outside an Arrow slice. Please account for non-null rows and
logical slice offsets, or cap the initial capacity and grow as needed.
Evidence: Calling unchanged base/head `spark_split_sql` with an 8,192-byte
scalar and `StringArray::from(vec![None::<&str>; 8192])` returns 8,192 nulls in
both cases, but `get_buffer_memory_size()` grows from 38,920 to 67,273,736
bytes. A second bounded probe constructs `[8 MiB string, "a,b"]`, slices to the
second row, and splits on comma. Both return `["a", "b"]`, while result buffer
memory grows from 5,132 to 8,388,687 bytes.
##########
native/spark-expr/src/string_funcs/split.rs:
##########
@@ -810,6 +991,361 @@ mod tests {
assert!(msg.contains("Invalid regex pattern '(unclosed'"), "{msg}");
}
+ #[test]
+ fn test_split_sql_empty_delimiter_scalar() {
+ let input = ColumnarValue::Scalar(ScalarValue::Utf8(Some("hello
world".to_string())));
+ let delimiter =
ColumnarValue::Scalar(ScalarValue::Utf8(Some("".to_string())));
+
+ let result = spark_split_sql(&[input, delimiter])
+ .unwrap()
+ .into_array(1)
+ .unwrap();
+ let list_array = result.as_any().downcast_ref::<ListArray>().unwrap();
+
+ assert_eq!(list_array.len(), 1);
+ let values = list_array.value(0);
+ let str_array = values.as_any().downcast_ref::<StringArray>().unwrap();
+
+ assert_eq!(str_array.len(), 1);
+ assert_eq!(str_array.value(0), "hello world");
+ }
+
+ #[test]
+ fn test_split_sql_empty_string_and_empty_delimiter_scalar() {
+ let input =
ColumnarValue::Scalar(ScalarValue::Utf8(Some("".to_string())));
+ let delimiter =
ColumnarValue::Scalar(ScalarValue::Utf8(Some("".to_string())));
+
+ let result = spark_split_sql(&[input, delimiter])
+ .unwrap()
+ .into_array(1)
+ .unwrap();
+ let list_array = result.as_any().downcast_ref::<ListArray>().unwrap();
+
+ assert_eq!(list_array.len(), 1);
+ let values = list_array.value(0);
+ let str_array = values.as_any().downcast_ref::<StringArray>().unwrap();
+
+ assert_eq!(str_array.len(), 1);
+ assert_eq!(str_array.value(0), "");
+ }
+
+ #[test]
+ fn test_split_sql_scalar_empty_delimiter_keeps_whole_string() {
+ // Spark semantics: an empty delimiter must NOT split into characters.
+ let args = vec![
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some("abc".to_string()))),
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some("".to_string()))),
+ ];
+ let result = spark_split_sql(&args).unwrap();
+
+ let list = match result {
+ ColumnarValue::Array(arr) => arr
+ .as_any()
+ .downcast_ref::<GenericListArray<i32>>()
+ .expect("expected ListArray")
+ .clone(),
+ ColumnarValue::Scalar(ScalarValue::List(list)) => (*list).clone(),
+ other => panic!("unexpected result: {:?}", other.data_type()),
+ };
+
+ let first = list.value(0);
+ let items = first
+ .as_any()
+ .downcast_ref::<GenericStringArray<i32>>()
+ .expect("expected Utf8 items");
+
+ assert_eq!(items.len(), 1, "empty delimiter must not split into
chars");
+ assert_eq!(items.value(0), "abc");
+ }
+
+ #[test]
+ fn test_split_sql_scalar_empty_delimiter_empty_string() {
+ // Empty input with an empty delimiter -> [""], matching the array
path.
+ let args = vec![
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some("".to_string()))),
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some("".to_string()))),
+ ];
+ let result = spark_split_sql(&args).unwrap();
+
+ let list = match result {
+ ColumnarValue::Array(arr) => arr
+ .as_any()
+ .downcast_ref::<GenericListArray<i32>>()
+ .expect("expected ListArray")
+ .clone(),
+ ColumnarValue::Scalar(ScalarValue::List(list)) => (*list).clone(),
+ other => panic!("unexpected result: {:?}", other.data_type()),
+ };
+
+ let first = list.value(0);
+ let items = first
+ .as_any()
+ .downcast_ref::<GenericStringArray<i32>>()
+ .expect("expected Utf8 items");
+
+ assert_eq!(items.len(), 1);
+ assert_eq!(items.value(0), "");
+ }
+
+ #[test]
+ fn test_split_sql_empty_delimiter_scalar_array_parity() {
+ // Scalar and array inputs must give identical results for the
+ // empty-delimiter case: the whole string as a single element.
+ let strings = vec!["abc", "", "hello world"];
+
+ let array_args = vec![
+ ColumnarValue::Array(Arc::new(StringArray::from(strings.clone()))),
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some("".to_string()))),
+ ];
+ let array_result = spark_split_sql(&array_args).unwrap();
+ let array_list = match array_result {
+ ColumnarValue::Array(arr) => arr
+ .as_any()
+ .downcast_ref::<GenericListArray<i32>>()
+ .expect("expected ListArray")
+ .clone(),
+ other => panic!("unexpected result: {:?}", other.data_type()),
+ };
+
+ for (row, s) in strings.iter().enumerate() {
+ let item = array_list.value(row);
+ let items = item
+ .as_any()
+ .downcast_ref::<GenericStringArray<i32>>()
+ .expect("expected Utf8 items");
+ assert_eq!(items.len(), 1, "row {}: expected single element", row);
+ assert_eq!(items.value(0), *s, "row {}", row);
+ }
+
+ let scalar_args = vec![
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some("abc".to_string()))),
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some("".to_string()))),
+ ];
+ let scalar_result = spark_split_sql(&scalar_args).unwrap();
+ let scalar_list = match scalar_result {
+ ColumnarValue::Array(arr) => arr
+ .as_any()
+ .downcast_ref::<GenericListArray<i32>>()
+ .expect("expected ListArray")
+ .clone(),
+ ColumnarValue::Scalar(ScalarValue::List(list)) => (*list).clone(),
+ other => panic!("unexpected result: {:?}", other.data_type()),
+ };
+
+ let first = scalar_list.value(0);
+ let items = first
+ .as_any()
+ .downcast_ref::<GenericStringArray<i32>>()
+ .expect("expected Utf8 items");
+ assert_eq!(items.len(), 1);
+ assert_eq!(items.value(0), "abc");
+ }
+
+ #[test]
+ fn test_split_sql_scalar_delimiter_still_splits_normally() {
+ // Guard: the empty-delimiter handling must not affect normal
delimiters.
+ let args = vec![
+
ColumnarValue::Scalar(ScalarValue::Utf8(Some("a,b,c".to_string()))),
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some(",".to_string()))),
+ ];
+ let result = spark_split_sql(&args).unwrap();
+
+ let list = match result {
+ ColumnarValue::Array(arr) => arr
+ .as_any()
+ .downcast_ref::<GenericListArray<i32>>()
+ .expect("expected ListArray")
+ .clone(),
+ ColumnarValue::Scalar(ScalarValue::List(list)) => (*list).clone(),
+ other => panic!("unexpected result: {:?}", other.data_type()),
+ };
+
+ let first = list.value(0);
+ let items = first
+ .as_any()
+ .downcast_ref::<GenericStringArray<i32>>()
+ .expect("expected Utf8 items");
+
+ assert_eq!(items.len(), 3);
+ assert_eq!(items.value(0), "a");
+ assert_eq!(items.value(1), "b");
+ assert_eq!(items.value(2), "c");
+ }
+
+ #[test]
+ fn test_split_sql_scalar_string_array_delimiter_with_empty_element() {
+ // (Scalar string, Array delimiter): an empty delimiter element must
+ // keep the whole string as one element, like the other branches.
+ let delimiters = StringArray::from(vec![Some(""), Some(",")]);
+ let args = vec![
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some("abc".to_string()))),
+ ColumnarValue::Array(Arc::new(delimiters)),
+ ];
+ let result = spark_split_sql(&args).unwrap();
+
+ let list = match result {
+ ColumnarValue::Array(arr) => arr
+ .as_any()
+ .downcast_ref::<GenericListArray<i32>>()
+ .expect("expected ListArray")
+ .clone(),
+ other => panic!("unexpected result: {:?}", other.data_type()),
+ };
+
+ assert_eq!(list.len(), 2);
+
+ let first = list.value(0);
+ let row0 = first
+ .as_any()
+ .downcast_ref::<GenericStringArray<i32>>()
+ .expect("expected Utf8 items");
+ assert_eq!(row0.len(), 1, "empty delimiter must not split into chars");
+ assert_eq!(row0.value(0), "abc");
+
+ let second = list.value(1);
+ let row1 = second
+ .as_any()
+ .downcast_ref::<GenericStringArray<i32>>()
+ .expect("expected Utf8 items");
+ assert_eq!(row1.len(), 1);
+ assert_eq!(row1.value(0), "abc");
+ }
+
+ #[test]
+ fn test_split_sql_scalar_item_field_non_nullable() {
+ // The scalar branch must produce the same List type as the array
+ // branches: a non-nullable "item" field.
+ let args = vec![
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some("abc".to_string()))),
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some(",".to_string()))),
+ ];
+ let result = spark_split_sql(&args).unwrap();
+
+ let list = match result {
+ ColumnarValue::Array(arr) => arr
+ .as_any()
+ .downcast_ref::<GenericListArray<i32>>()
+ .expect("expected ListArray")
+ .clone(),
+ ColumnarValue::Scalar(ScalarValue::List(list)) => (*list).clone(),
+ other => panic!("unexpected result: {:?}", other.data_type()),
+ };
+
+ let item_field = match list.data_type() {
+ DataType::List(field) | DataType::LargeList(field) =>
field.clone(),
+ other => panic!("expected List type, got {:?}", other),
+ };
+ assert!(
+ !item_field.is_nullable(),
+ "scalar split_sql must keep the non-nullable item field"
+ );
+
+ // parity with the array path
+ let array_args = vec![
+ ColumnarValue::Array(Arc::new(StringArray::from(vec!["abc"]))),
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some(",".to_string()))),
+ ];
+ let array_result = spark_split_sql(&array_args).unwrap();
+ let array_list = match array_result {
+ ColumnarValue::Array(arr) => arr
+ .as_any()
+ .downcast_ref::<GenericListArray<i32>>()
+ .expect("expected ListArray")
+ .clone(),
+ other => panic!("unexpected result: {:?}", other.data_type()),
+ };
+
+ assert_eq!(
+ list.data_type(),
+ array_list.data_type(),
+ "scalar and array split_sql must return the same List type"
+ );
+ }
+
+ fn scalar_split_to_vec(s: &str, pattern: &str, limit: i32) -> Vec<String> {
+ let args = vec![
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some(s.to_string()))),
+
ColumnarValue::Scalar(ScalarValue::Utf8(Some(pattern.to_string()))),
+ ColumnarValue::Scalar(ScalarValue::Int32(Some(limit))),
+ ];
+ match spark_split(&args).unwrap() {
Review Comment:
[P1] [P1] Repair the test and benchmark callers before merging
The Rust test target no longer compiles. This new call omits the required
`&PatternCache`, while retained tests still call the deleted `split_string` and
`split_sql_string` helpers. The benchmark also omits the cache at lines 78 and
97. These failures block the Rust CI gates and prevent running the advertised
tests and benchmarks. Please migrate retained tests to the public entry points,
update the old limit-zero expectation, and restore one persistent cache per
benchmark input.
Evidence: `cd native && cargo test --locked --offline -p
datafusion-comet-spark-expr --lib string_funcs::split --no-run` exits 101 with
four E0425 errors at split.rs:793/882/888/894 and E0061 at :1271. Compiling the
unchanged benchmark against the exact-head function independently produces
E0061 at benches/split.rs:78 and :97. The isolated exact-base split/cache
module compiles and passes all 19 tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
