| // 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::{ | |
| Array, ArrayBuilder, ArrayRef, GenericListArray, GenericStringArray, GenericStringBuilder, | |
| ListArray, NullBufferBuilder, OffsetSizeTrait, | |
| }; | |
| use arrow::buffer::OffsetBuffer; | |
| use arrow::datatypes::{DataType, Field}; | |
| use datafusion::common::{ | |
| cast::as_generic_string_array, exec_err, DataFusionError, Result as DataFusionResult, | |
| ScalarValue, | |
| }; | |
| use datafusion::logical_expr::ColumnarValue; | |
| use regex::Regex; | |
| use std::sync::Arc; | |
| use super::pattern_cache::PatternCache; | |
| /// Spark-compatible split function | |
| /// Splits a string around matches of a regex pattern with optional limit | |
| /// | |
| /// Arguments: | |
| /// - string: The string to split | |
| /// - pattern: The regex pattern to split on | |
| /// - limit (optional): Controls the number of splits | |
| /// - limit > 0: At most limit-1 splits, array length <= limit | |
| /// - limit = 0: As many splits as possible, trailing empty strings removed | |
| /// - limit < 0: As many splits as possible, trailing empty strings kept | |
| pub fn spark_split( | |
| args: &[ColumnarValue], | |
| regex_cache: &PatternCache, | |
| ) -> DataFusionResult<ColumnarValue> { | |
| if args.len() < 2 || args.len() > 3 { | |
| return exec_err!( | |
| "split expects 2 or 3 arguments (string, pattern, [limit]), got {}", | |
| args.len() | |
| ); | |
| } | |
| // Get limit parameter (default to -1 if not provided) | |
| let limit = if args.len() == 3 { | |
| match &args[2] { | |
| ColumnarValue::Scalar(ScalarValue::Int32(Some(l))) => *l, | |
| ColumnarValue::Scalar(ScalarValue::Int32(None)) => { | |
| // NULL limit, return NULL | |
| return Ok(ColumnarValue::Scalar(ScalarValue::Null)); | |
| } | |
| _ => { | |
| return exec_err!("split limit argument must be an Int32 scalar"); | |
| } | |
| } | |
| } else { | |
| -1 | |
| }; | |
| match (&args[0], &args[1]) { | |
| (ColumnarValue::Array(string_array), ColumnarValue::Scalar(ScalarValue::Utf8(pattern))) | |
| | ( | |
| ColumnarValue::Array(string_array), | |
| ColumnarValue::Scalar(ScalarValue::LargeUtf8(pattern)), | |
| ) => { | |
| if pattern.is_none() { | |
| // NULL pattern returns NULL | |
| let null_array = new_null_list_array(string_array.len()); | |
| return Ok(ColumnarValue::Array(null_array)); | |
| } | |
| let pattern_str = pattern.as_ref().unwrap(); | |
| split_array(string_array.as_ref(), pattern_str, limit, regex_cache) | |
| } | |
| (ColumnarValue::Scalar(ScalarValue::Utf8(string)), ColumnarValue::Scalar(pattern_val)) | |
| | ( | |
| ColumnarValue::Scalar(ScalarValue::LargeUtf8(string)), | |
| ColumnarValue::Scalar(pattern_val), | |
| ) => { | |
| if string.is_none() { | |
| return Ok(ColumnarValue::Scalar(new_null_list_scalar())); | |
| } | |
| let pattern_str = match pattern_val { | |
| ScalarValue::Utf8(Some(p)) | ScalarValue::LargeUtf8(Some(p)) => p, | |
| ScalarValue::Utf8(None) | ScalarValue::LargeUtf8(None) => { | |
| return Ok(ColumnarValue::Scalar(new_null_list_scalar())); | |
| } | |
| _ => { | |
| return exec_err!("split pattern must be a string"); | |
| } | |
| }; | |
| 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)); | |
| Ok(ColumnarValue::Scalar(ScalarValue::List(Arc::new( | |
| list_array, | |
| )))) | |
| } | |
| _ => exec_err!("split expects (array, scalar) or (scalar, scalar) arguments"), | |
| } | |
| } | |
| /// Spark-compatible StringSplitSQL function. | |
| /// Splits a string around literal delimiter matches and keeps trailing empty strings. | |
| pub fn spark_split_sql(args: &[ColumnarValue]) -> DataFusionResult<ColumnarValue> { | |
| if args.len() != 2 { | |
| return exec_err!( | |
| "split_sql expects 2 arguments (string, delimiter), got {}", | |
| args.len() | |
| ); | |
| } | |
| match (&args[0], &args[1]) { | |
| (ColumnarValue::Array(string_array), ColumnarValue::Scalar(delimiter)) => { | |
| let delimiter = match delimiter { | |
| ScalarValue::Utf8(Some(d)) | ScalarValue::LargeUtf8(Some(d)) => d, | |
| ScalarValue::Utf8(None) | ScalarValue::LargeUtf8(None) => { | |
| return Ok(ColumnarValue::Array(new_null_list_array( | |
| string_array.len(), | |
| ))); | |
| } | |
| _ => return exec_err!("split_sql delimiter must be a string"), | |
| }; | |
| split_sql_array_scalar(string_array.as_ref(), delimiter) | |
| } | |
| (ColumnarValue::Array(string_array), ColumnarValue::Array(delimiter_array)) => { | |
| split_sql_array_array(string_array.as_ref(), delimiter_array.as_ref()) | |
| } | |
| ( | |
| ColumnarValue::Scalar(ScalarValue::Utf8(string)), | |
| ColumnarValue::Array(delimiter_array), | |
| ) => split_sql_scalar_array::<i32>(string.as_deref(), delimiter_array.as_ref()), | |
| ( | |
| ColumnarValue::Scalar(ScalarValue::LargeUtf8(string)), | |
| ColumnarValue::Array(delimiter_array), | |
| ) => split_sql_scalar_array::<i64>(string.as_deref(), delimiter_array.as_ref()), | |
| (ColumnarValue::Scalar(ScalarValue::Utf8(string)), ColumnarValue::Scalar(delimiter)) | |
| | ( | |
| ColumnarValue::Scalar(ScalarValue::LargeUtf8(string)), | |
| ColumnarValue::Scalar(delimiter), | |
| ) => { | |
| if string.is_none() { | |
| return Ok(ColumnarValue::Scalar(new_null_list_scalar())); | |
| } | |
| let delimiter = match delimiter { | |
| ScalarValue::Utf8(Some(d)) | ScalarValue::LargeUtf8(Some(d)) => d, | |
| ScalarValue::Utf8(None) | ScalarValue::LargeUtf8(None) => { | |
| return Ok(ColumnarValue::Scalar(new_null_list_scalar())); | |
| } | |
| _ => return exec_err!("split_sql delimiter must be a string"), | |
| }; | |
| let result = split_sql_string(string.as_ref().unwrap(), delimiter); | |
| let string_array = GenericStringArray::<i32>::from(result); | |
| let list_array = create_list_array(Arc::new(string_array)); | |
| Ok(ColumnarValue::Scalar(ScalarValue::List(Arc::new( | |
| list_array, | |
| )))) | |
| } | |
| _ => exec_err!("split_sql expects string arguments"), | |
| } | |
| } | |
| fn split_array( | |
| string_array: &dyn arrow::array::Array, | |
| pattern: &str, | |
| limit: i32, | |
| regex_cache: &PatternCache, | |
| ) -> DataFusionResult<ColumnarValue> { | |
| // The pattern is a plan-time literal, so the cache makes this compile a one-time cost | |
| // for the expression instead of a per-batch cost. | |
| let regex = regex_cache.get_or_compile(pattern).map_err(|e| { | |
| DataFusionError::Execution(format!("Invalid regex pattern '{}': {}", pattern, e)) | |
| })?; | |
| match string_array.data_type() { | |
| DataType::Utf8 => { | |
| split_generic::<i32>(as_generic_string_array::<i32>(string_array)?, ®ex, limit) | |
| } | |
| DataType::LargeUtf8 => { | |
| split_generic::<i64>(as_generic_string_array::<i64>(string_array)?, ®ex, limit) | |
| } | |
| _ => exec_err!( | |
| "split expects Utf8 or LargeUtf8 string array, got {:?}", | |
| string_array.data_type() | |
| ), | |
| } | |
| } | |
| fn split_sql_array_scalar( | |
| string_array: &dyn arrow::array::Array, | |
| delimiter: &str, | |
| ) -> DataFusionResult<ColumnarValue> { | |
| match string_array.data_type() { | |
| DataType::Utf8 => split_sql_generic_scalar::<i32>( | |
| as_generic_string_array::<i32>(string_array)?, | |
| delimiter, | |
| ), | |
| DataType::LargeUtf8 => split_sql_generic_scalar::<i64>( | |
| as_generic_string_array::<i64>(string_array)?, | |
| delimiter, | |
| ), | |
| _ => exec_err!( | |
| "split_sql expects Utf8 or LargeUtf8 string array, got {:?}", | |
| string_array.data_type() | |
| ), | |
| } | |
| } | |
| fn split_sql_array_array( | |
| string_array: &dyn arrow::array::Array, | |
| delimiter_array: &dyn arrow::array::Array, | |
| ) -> DataFusionResult<ColumnarValue> { | |
| if string_array.len() != delimiter_array.len() { | |
| return exec_err!( | |
| "split_sql string and delimiter arrays must have the same length, got {} and {}", | |
| string_array.len(), | |
| delimiter_array.len() | |
| ); | |
| } | |
| match (string_array.data_type(), delimiter_array.data_type()) { | |
| (DataType::Utf8, DataType::Utf8) => split_sql_generic_array::<i32, i32>( | |
| as_generic_string_array::<i32>(string_array)?, | |
| as_generic_string_array::<i32>(delimiter_array)?, | |
| ), | |
| (DataType::Utf8, DataType::LargeUtf8) => split_sql_generic_array::<i32, i64>( | |
| as_generic_string_array::<i32>(string_array)?, | |
| as_generic_string_array::<i64>(delimiter_array)?, | |
| ), | |
| (DataType::LargeUtf8, DataType::Utf8) => split_sql_generic_array::<i64, i32>( | |
| as_generic_string_array::<i64>(string_array)?, | |
| as_generic_string_array::<i32>(delimiter_array)?, | |
| ), | |
| (DataType::LargeUtf8, DataType::LargeUtf8) => split_sql_generic_array::<i64, i64>( | |
| as_generic_string_array::<i64>(string_array)?, | |
| as_generic_string_array::<i64>(delimiter_array)?, | |
| ), | |
| _ => exec_err!( | |
| "split_sql expects Utf8 or LargeUtf8 string arrays, got {:?} and {:?}", | |
| string_array.data_type(), | |
| delimiter_array.data_type() | |
| ), | |
| } | |
| } | |
| fn split_sql_scalar_array<O: OffsetSizeTrait>( | |
| string: Option<&str>, | |
| delimiter_array: &dyn arrow::array::Array, | |
| ) -> DataFusionResult<ColumnarValue> { | |
| let Some(string) = string else { | |
| return Ok(ColumnarValue::Array(new_null_list_array_with_offset::<O>( | |
| delimiter_array.len(), | |
| ))); | |
| }; | |
| match delimiter_array.data_type() { | |
| DataType::Utf8 => split_sql_generic_scalar_array::<O, i32>( | |
| string, | |
| as_generic_string_array::<i32>(delimiter_array)?, | |
| ), | |
| DataType::LargeUtf8 => split_sql_generic_scalar_array::<O, i64>( | |
| string, | |
| as_generic_string_array::<i64>(delimiter_array)?, | |
| ), | |
| _ => exec_err!( | |
| "split_sql expects Utf8 or LargeUtf8 delimiter array, got {:?}", | |
| delimiter_array.data_type() | |
| ), | |
| } | |
| } | |
| fn split_generic<O: OffsetSizeTrait>( | |
| string_array: &GenericStringArray<O>, | |
| regex: &Regex, | |
| limit: i32, | |
| ) -> DataFusionResult<ColumnarValue> { | |
| let len = string_array.len(); | |
| let mut offsets: Vec<O> = Vec::with_capacity(len + 1); | |
| let mut values_builder = GenericStringBuilder::<O>::new(); | |
| offsets.push(O::usize_as(0)); | |
| // Bulk-NULL: output null mask equals input's, so reuse it instead of | |
| // tracking per-row in a NullBufferBuilder. Null rows contribute no parts | |
| // (offset does not advance) and the cloned NullBuffer marks them. | |
| for i in 0..len { | |
| if !string_array.is_null(i) { | |
| let s = string_array.value(i); | |
| push_split_parts(s, regex, limit, &mut values_builder); | |
| } | |
| offsets.push(O::usize_as(values_builder.len())); | |
| } | |
| let values_array = Arc::new(values_builder.finish()) as ArrayRef; | |
| let item_type = if O::IS_LARGE { | |
| DataType::LargeUtf8 | |
| } else { | |
| DataType::Utf8 | |
| }; | |
| let field = Arc::new(Field::new("item", item_type, false)); | |
| let list_array = GenericListArray::<O>::new( | |
| field, | |
| OffsetBuffer::new(offsets.into()), | |
| values_array, | |
| string_array.nulls().cloned(), | |
| ); | |
| Ok(ColumnarValue::Array(Arc::new(list_array))) | |
| } | |
| fn split_sql_generic_scalar<O: OffsetSizeTrait>( | |
| string_array: &GenericStringArray<O>, | |
| delimiter: &str, | |
| ) -> DataFusionResult<ColumnarValue> { | |
| let len = string_array.len(); | |
| let mut offsets: Vec<O> = Vec::with_capacity(len + 1); | |
| let mut values_builder = GenericStringBuilder::<O>::new(); | |
| offsets.push(O::usize_as(0)); | |
| for i in 0..len { | |
| if !string_array.is_null(i) { | |
| push_split_sql_parts(string_array.value(i), delimiter, &mut values_builder); | |
| } | |
| offsets.push(O::usize_as(values_builder.len())); | |
| } | |
| let values_array = Arc::new(values_builder.finish()) as ArrayRef; | |
| let item_type = if O::IS_LARGE { | |
| DataType::LargeUtf8 | |
| } else { | |
| DataType::Utf8 | |
| }; | |
| let field = Arc::new(Field::new("item", item_type, false)); | |
| let list_array = GenericListArray::<O>::new( | |
| field, | |
| OffsetBuffer::new(offsets.into()), | |
| values_array, | |
| string_array.nulls().cloned(), | |
| ); | |
| Ok(ColumnarValue::Array(Arc::new(list_array))) | |
| } | |
| fn split_sql_generic_scalar_array<O: OffsetSizeTrait, D: OffsetSizeTrait>( | |
| string: &str, | |
| delimiter_array: &GenericStringArray<D>, | |
| ) -> 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 mut nulls = NullBufferBuilder::new(len); | |
| offsets.push(O::usize_as(0)); | |
| for i in 0..len { | |
| if delimiter_array.is_null(i) { | |
| nulls.append_null(); | |
| } else { | |
| push_split_sql_parts(string, delimiter_array.value(i), &mut values_builder); | |
| nulls.append_non_null(); | |
| } | |
| offsets.push(O::usize_as(values_builder.len())); | |
| } | |
| let values_array = Arc::new(values_builder.finish()) as ArrayRef; | |
| let item_type = if O::IS_LARGE { | |
| DataType::LargeUtf8 | |
| } else { | |
| DataType::Utf8 | |
| }; | |
| let field = Arc::new(Field::new("item", item_type, false)); | |
| let list_array = GenericListArray::<O>::new( | |
| field, | |
| OffsetBuffer::new(offsets.into()), | |
| values_array, | |
| nulls.finish(), | |
| ); | |
| Ok(ColumnarValue::Array(Arc::new(list_array))) | |
| } | |
| fn split_sql_generic_array<O: OffsetSizeTrait, D: OffsetSizeTrait>( | |
| string_array: &GenericStringArray<O>, | |
| delimiter_array: &GenericStringArray<D>, | |
| ) -> DataFusionResult<ColumnarValue> { | |
| let len = string_array.len(); | |
| let mut offsets: Vec<O> = Vec::with_capacity(len + 1); | |
| let mut values_builder = GenericStringBuilder::<O>::new(); | |
| let mut nulls = NullBufferBuilder::new(len); | |
| offsets.push(O::usize_as(0)); | |
| for i in 0..len { | |
| if string_array.is_null(i) || delimiter_array.is_null(i) { | |
| nulls.append_null(); | |
| } else { | |
| push_split_sql_parts( | |
| string_array.value(i), | |
| delimiter_array.value(i), | |
| &mut values_builder, | |
| ); | |
| nulls.append_non_null(); | |
| } | |
| offsets.push(O::usize_as(values_builder.len())); | |
| } | |
| let values_array = Arc::new(values_builder.finish()) as ArrayRef; | |
| let item_type = if O::IS_LARGE { | |
| DataType::LargeUtf8 | |
| } else { | |
| DataType::Utf8 | |
| }; | |
| let field = Arc::new(Field::new("item", item_type, false)); | |
| let list_array = GenericListArray::<O>::new( | |
| field, | |
| OffsetBuffer::new(offsets.into()), | |
| values_array, | |
| nulls.finish(), | |
| ); | |
| Ok(ColumnarValue::Array(Arc::new(list_array))) | |
| } | |
| /// Push the splits of `string` into `builder`. Avoids materializing an | |
| /// intermediate `Vec<String>` — appends each `&str` slice from the regex | |
| /// iterator directly (the builder copies into its own buffer). | |
| fn push_split_parts<O: OffsetSizeTrait>( | |
| string: &str, | |
| regex: &Regex, | |
| limit: i32, | |
| builder: &mut GenericStringBuilder<O>, | |
| ) { | |
| if limit == 0 { | |
| // limit = 0: split all, drop trailing empties. Need to know the end | |
| // before pushing, so collect borrowed slices first (no string copies). | |
| let mut parts: Vec<&str> = regex.split(string).collect(); | |
| while parts.last().is_some_and(|s| s.is_empty()) { | |
| parts.pop(); | |
| } | |
| if parts.is_empty() { | |
| builder.append_value(""); | |
| } else { | |
| for p in parts { | |
| builder.append_value(p); | |
| } | |
| } | |
| } else if limit > 0 { | |
| // limit > 0: at most limit-1 splits. | |
| let mut last_end = 0; | |
| let cap = (limit - 1) as usize; | |
| for (count, mat) in regex.find_iter(string).enumerate() { | |
| if count >= cap { | |
| break; | |
| } | |
| builder.append_value(&string[last_end..mat.start()]); | |
| last_end = mat.end(); | |
| } | |
| builder.append_value(&string[last_end..]); | |
| } else { | |
| // limit < 0: split all, keep trailing empties. | |
| for p in regex.split(string) { | |
| builder.append_value(p); | |
| } | |
| } | |
| } | |
| fn push_split_sql_parts<O: OffsetSizeTrait>( | |
| string: &str, | |
| delimiter: &str, | |
| builder: &mut GenericStringBuilder<O>, | |
| ) { | |
| if delimiter.is_empty() { | |
| builder.append_value(string); | |
| } else { | |
| for p in string.split(delimiter) { | |
| builder.append_value(p); | |
| } | |
| } | |
| } | |
| fn split_string( | |
| string: &str, | |
| pattern: &str, | |
| limit: i32, | |
| regex_cache: &PatternCache, | |
| ) -> DataFusionResult<Vec<String>> { | |
| let regex = regex_cache.get_or_compile(pattern).map_err(|e| { | |
| DataFusionError::Execution(format!("Invalid regex pattern '{}': {}", pattern, e)) | |
| })?; | |
| Ok(split_with_regex(string, ®ex, limit)) | |
| } | |
| fn split_with_regex(string: &str, regex: &Regex, limit: i32) -> Vec<String> { | |
| if limit == 0 { | |
| // limit = 0: split as many times as possible, discard trailing empty strings | |
| let mut parts: Vec<String> = regex.split(string).map(|s| s.to_string()).collect(); | |
| // Remove trailing empty strings | |
| while parts.last().is_some_and(|s| s.is_empty()) { | |
| parts.pop(); | |
| } | |
| if parts.is_empty() { | |
| vec!["".to_string()] | |
| } else { | |
| parts | |
| } | |
| } else if limit > 0 { | |
| // limit > 0: at most limit-1 splits (array length <= limit) | |
| let mut parts: Vec<String> = Vec::new(); | |
| let mut last_end = 0; | |
| for (count, mat) in regex.find_iter(string).enumerate() { | |
| if count >= (limit - 1) as usize { | |
| break; | |
| } | |
| parts.push(string[last_end..mat.start()].to_string()); | |
| last_end = mat.end(); | |
| } | |
| // Add the remaining string | |
| parts.push(string[last_end..].to_string()); | |
| parts | |
| } else { | |
| // limit < 0: split as many times as possible, keep trailing empty strings | |
| regex.split(string).map(|s| s.to_string()).collect() | |
| } | |
| } | |
| fn split_sql_string(string: &str, delimiter: &str) -> Vec<String> { | |
| if delimiter.is_empty() { | |
| vec![string.to_string()] | |
| } else { | |
| string.split(delimiter).map(|s| s.to_string()).collect() | |
| } | |
| } | |
| fn create_list_array(values: ArrayRef) -> ListArray { | |
| let field = Arc::new(Field::new("item", DataType::Utf8, false)); | |
| let offsets = vec![0i32, values.len() as i32]; | |
| ListArray::new( | |
| field, | |
| arrow::buffer::OffsetBuffer::new(offsets.into()), | |
| values, | |
| None, | |
| ) | |
| } | |
| fn new_null_list_array(len: usize) -> ArrayRef { | |
| Arc::new(new_null_list_array_value(len)) | |
| } | |
| fn new_null_list_scalar() -> ScalarValue { | |
| ScalarValue::List(Arc::new(new_null_list_array_value(1))) | |
| } | |
| fn new_null_list_array_with_offset<O: OffsetSizeTrait>(len: usize) -> ArrayRef { | |
| let item_type = if O::IS_LARGE { | |
| DataType::LargeUtf8 | |
| } else { | |
| DataType::Utf8 | |
| }; | |
| let field = Arc::new(Field::new("item", item_type, false)); | |
| let values = Arc::new(GenericStringArray::<O>::from(Vec::<String>::new())) as ArrayRef; | |
| let offsets = vec![O::usize_as(0); len + 1]; | |
| let nulls = arrow::buffer::NullBuffer::new_null(len); | |
| Arc::new(GenericListArray::<O>::new( | |
| field, | |
| arrow::buffer::OffsetBuffer::new(offsets.into()), | |
| values, | |
| Some(nulls), | |
| )) | |
| } | |
| fn new_null_list_array_value(len: usize) -> ListArray { | |
| let field = Arc::new(Field::new("item", DataType::Utf8, false)); | |
| let values = Arc::new(GenericStringArray::<i32>::from(Vec::<String>::new())) as ArrayRef; | |
| let offsets = vec![0i32; len + 1]; | |
| let nulls = arrow::buffer::NullBuffer::new_null(len); | |
| ListArray::new( | |
| field, | |
| arrow::buffer::OffsetBuffer::new(offsets.into()), | |
| values, | |
| Some(nulls), | |
| ) | |
| } | |
| #[cfg(test)] | |
| mod tests { | |
| use super::*; | |
| use arrow::array::StringArray; | |
| fn run_split(args: &[ColumnarValue]) -> DataFusionResult<ColumnarValue> { | |
| spark_split(args, &PatternCache::new()) | |
| } | |
| fn run_split_string(string: &str, pattern: &str, limit: i32) -> DataFusionResult<Vec<String>> { | |
| split_string(string, pattern, limit, &PatternCache::new()) | |
| } | |
| #[test] | |
| fn test_split_basic() { | |
| let string_array = Arc::new(StringArray::from(vec!["a,b,c", "x,y,z"])) as ArrayRef; | |
| let pattern = ColumnarValue::Scalar(ScalarValue::Utf8(Some(",".to_string()))); | |
| let args = vec![ColumnarValue::Array(string_array), pattern]; | |
| let result = run_split(&args).unwrap(); | |
| // Should produce [["a", "b", "c"], ["x", "y", "z"]] | |
| assert!(matches!(result, ColumnarValue::Array(_))); | |
| } | |
| #[test] | |
| fn test_split_with_limit() { | |
| let string_array = Arc::new(StringArray::from(vec!["a,b,c,d"])) as ArrayRef; | |
| let pattern = ColumnarValue::Scalar(ScalarValue::Utf8(Some(",".to_string()))); | |
| let limit = ColumnarValue::Scalar(ScalarValue::Int32(Some(2))); | |
| let args = vec![ColumnarValue::Array(string_array), pattern, limit]; | |
| let result = run_split(&args).unwrap(); | |
| // Should produce [["a", "b,c,d"]] | |
| assert!(matches!(result, ColumnarValue::Array(_))); | |
| } | |
| #[test] | |
| fn test_split_regex() { | |
| let parts = run_split_string("foo123bar456baz", r"\d+", -1).unwrap(); | |
| assert_eq!(parts, vec!["foo", "bar", "baz"]); | |
| } | |
| #[test] | |
| fn test_split_limit_positive() { | |
| let parts = run_split_string("a,b,c,d,e", ",", 3).unwrap(); | |
| assert_eq!(parts, vec!["a", "b", "c,d,e"]); | |
| } | |
| #[test] | |
| fn test_split_limit_zero() { | |
| let parts = run_split_string("a,b,c,,", ",", 0).unwrap(); | |
| assert_eq!(parts, vec!["a", "b", "c"]); | |
| } | |
| #[test] | |
| fn test_split_limit_negative() { | |
| let parts = run_split_string("a,b,c,,", ",", -1).unwrap(); | |
| assert_eq!(parts, vec!["a", "b", "c", "", ""]); | |
| } | |
| #[test] | |
| fn test_split_with_nulls() { | |
| // Test that NULL inputs produce NULL outputs (not empty arrays) | |
| let string_array = Arc::new(StringArray::from(vec![ | |
| Some("a,b,c"), | |
| None, | |
| Some("x,y"), | |
| None, | |
| ])) as ArrayRef; | |
| let pattern = ColumnarValue::Scalar(ScalarValue::Utf8(Some(",".to_string()))); | |
| let args = vec![ColumnarValue::Array(string_array), pattern]; | |
| let result = run_split(&args).unwrap(); | |
| match result { | |
| ColumnarValue::Array(arr) => { | |
| let list_array = arr.as_any().downcast_ref::<ListArray>().unwrap(); | |
| assert_eq!(list_array.len(), 4); | |
| // First row: valid ["a", "b", "c"] | |
| assert!(!list_array.is_null(0)); | |
| // Second row: NULL | |
| assert!(list_array.is_null(1)); | |
| // Third row: valid ["x", "y"] | |
| assert!(!list_array.is_null(2)); | |
| // Fourth row: NULL | |
| assert!(list_array.is_null(3)); | |
| } | |
| _ => panic!("Expected Array result"), | |
| } | |
| } | |
| #[test] | |
| fn test_split_empty_string() { | |
| // Test that empty string input produces array with single empty string | |
| let parts = run_split_string("", ",", -1).unwrap(); | |
| assert_eq!(parts, vec![""]); | |
| } | |
| #[test] | |
| fn test_split_sql_literal_delimiter() { | |
| let parts = split_sql_string("a.b.", "."); | |
| assert_eq!(parts, vec!["a", "b", ""]); | |
| } | |
| #[test] | |
| fn test_split_sql_empty_delimiter() { | |
| let parts = split_sql_string("abc", ""); | |
| assert_eq!(parts, vec!["abc"]); | |
| } | |
| #[test] | |
| fn test_split_sql_keeps_regex_chars_literal() { | |
| let parts = split_sql_string("a.b.c", "."); | |
| assert_eq!(parts, vec!["a", "b", "c"]); | |
| } | |
| #[test] | |
| fn test_split_sql_scalar_nulls_return_typed_null_list() { | |
| let delimiter = ColumnarValue::Scalar(ScalarValue::Utf8(Some(",".to_string()))); | |
| let result = spark_split_sql(&[ | |
| ColumnarValue::Scalar(ScalarValue::Utf8(None)), | |
| delimiter.clone(), | |
| ]) | |
| .unwrap(); | |
| assert_null_list_scalar(result); | |
| let result = spark_split_sql(&[ | |
| ColumnarValue::Scalar(ScalarValue::Utf8(Some("a,b".to_string()))), | |
| ColumnarValue::Scalar(ScalarValue::Utf8(None)), | |
| ]) | |
| .unwrap(); | |
| assert_null_list_scalar(result); | |
| } | |
| #[test] | |
| fn test_split_sql_scalar_string_array_delimiter() { | |
| let delimiter_array = | |
| Arc::new(StringArray::from(vec![Some("||"), Some("."), None])) as ArrayRef; | |
| let result = spark_split_sql(&[ | |
| ColumnarValue::Scalar(ScalarValue::Utf8(Some("a||b||".to_string()))), | |
| ColumnarValue::Array(delimiter_array), | |
| ]) | |
| .unwrap(); | |
| match result { | |
| ColumnarValue::Array(arr) => { | |
| let list_array = arr.as_any().downcast_ref::<ListArray>().unwrap(); | |
| assert_eq!(list_array.len(), 3); | |
| assert_list_value(list_array, 0, &["a", "b", ""]); | |
| assert_list_value(list_array, 1, &["a||b||"]); | |
| assert!(list_array.is_null(2)); | |
| } | |
| _ => panic!("Expected Array result"), | |
| } | |
| } | |
| #[test] | |
| fn test_split_sql_null_scalar_string_array_delimiter() { | |
| let delimiter_array = Arc::new(StringArray::from(vec![Some(","), Some(".")])) as ArrayRef; | |
| let result = spark_split_sql(&[ | |
| ColumnarValue::Scalar(ScalarValue::Utf8(None)), | |
| ColumnarValue::Array(delimiter_array), | |
| ]) | |
| .unwrap(); | |
| match result { | |
| ColumnarValue::Array(arr) => { | |
| let list_array = arr.as_any().downcast_ref::<ListArray>().unwrap(); | |
| assert_eq!(list_array.len(), 2); | |
| assert!(list_array.is_null(0)); | |
| assert!(list_array.is_null(1)); | |
| } | |
| _ => panic!("Expected Array result"), | |
| } | |
| } | |
| /// One expression evaluates many batches; the pattern must compile once and results | |
| /// must stay correct on every batch. | |
| #[test] | |
| fn test_split_compiles_regex_once_across_batches() { | |
| let cache = PatternCache::new(); | |
| let pattern = ColumnarValue::Scalar(ScalarValue::Utf8(Some(r"\d+".to_string()))); | |
| for batch in 0..4 { | |
| let input = format!("foo{batch}bar{batch}baz"); | |
| let string_array = Arc::new(StringArray::from(vec![input.as_str()])) as ArrayRef; | |
| let args = vec![ColumnarValue::Array(string_array), pattern.clone()]; | |
| let result = spark_split(&args, &cache).unwrap(); | |
| match result { | |
| ColumnarValue::Array(arr) => { | |
| let list_array = arr.as_any().downcast_ref::<ListArray>().unwrap(); | |
| assert_list_value(list_array, 0, &["foo", "bar", "baz"]); | |
| } | |
| _ => panic!("Expected Array result"), | |
| } | |
| } | |
| assert_eq!(cache.compile_count(), 1); | |
| } | |
| #[test] | |
| fn test_split_invalid_pattern_errors_at_evaluation() { | |
| let string_array = Arc::new(StringArray::from(vec!["abc"])) as ArrayRef; | |
| let pattern = ColumnarValue::Scalar(ScalarValue::Utf8(Some("(unclosed".to_string()))); | |
| let args = vec![ColumnarValue::Array(string_array), pattern]; | |
| let err = run_split(&args).err().unwrap(); | |
| let msg = err.to_string(); | |
| assert!(msg.contains("Invalid regex pattern '(unclosed'"), "{msg}"); | |
| } | |
| fn assert_list_value(list_array: &ListArray, row: usize, expected: &[&str]) { | |
| let value = list_array.value(row); | |
| let strings = value.as_any().downcast_ref::<StringArray>().unwrap(); | |
| let actual = strings.iter().collect::<Vec<_>>(); | |
| let expected = expected.iter().map(|s| Some(*s)).collect::<Vec<_>>(); | |
| assert_eq!(actual, expected); | |
| } | |
| fn assert_null_list_scalar(result: ColumnarValue) { | |
| match result { | |
| ColumnarValue::Scalar(ScalarValue::List(array)) => { | |
| assert_eq!(array.len(), 1); | |
| assert!(array.is_null(0)); | |
| } | |
| _ => panic!("Expected typed null list scalar, got {result:?}"), | |
| } | |
| } | |
| } |