Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions native/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions native/spark-expr/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ arrow = { workspace = true }
chrono = { workspace = true }
datafusion = { workspace = true }
chrono-tz = { workspace = true }
memchr = "2.7"
num = { workspace = true }
regex = { workspace = true }
serde_json = "1.0"
Expand Down
5 changes: 3 additions & 2 deletions native/spark-expr/src/comet_scalar_funcs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,8 @@ use crate::math_funcs::modulo_expr::spark_modulo;
use crate::{
spark_array_repeat, spark_ceil, spark_decimal_div, spark_decimal_integral_div, spark_floor,
spark_isnan, spark_lpad, spark_make_decimal, spark_read_side_padding, spark_round, spark_rpad,
spark_unhex, spark_unscaled_value, EvalMode, SparkBitwiseCount, SparkDateTrunc, SparkSizeFunc,
SparkStringSpace,
spark_unhex, spark_unscaled_value, EvalMode, SparkBitwiseCount, SparkContains, SparkDateTrunc,
SparkSizeFunc, SparkStringSpace,
};
use arrow::datatypes::DataType;
use datafusion::common::{DataFusionError, Result as DataFusionResult};
Expand Down Expand Up @@ -192,6 +192,7 @@ pub fn create_comet_physical_fun_with_eval_mode(
fn all_scalar_functions() -> Vec<Arc<ScalarUDF>> {
vec![
Arc::new(ScalarUDF::new_from_impl(SparkBitwiseCount::default())),
Arc::new(ScalarUDF::new_from_impl(SparkContains::default())),
Arc::new(ScalarUDF::new_from_impl(SparkDateTrunc::default())),
Arc::new(ScalarUDF::new_from_impl(SparkStringSpace::default())),
Arc::new(ScalarUDF::new_from_impl(SparkSizeFunc::default())),
Expand Down
282 changes: 282 additions & 0 deletions native/spark-expr/src/string_funcs/contains.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,282 @@
// 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.

//! Optimized `contains` string function for Spark compatibility.
//!
//! This implementation is optimized for the common case where the pattern
//! (second argument) is a scalar value. In this case, we use `memchr::memmem::Finder`
//! which is SIMD-optimized and reuses a single finder instance across all rows.
//!
//! The DataFusion built-in `contains` function uses `make_scalar_function` which
//! expands scalar values to arrays, losing the performance benefit of the optimized
//! scalar path in arrow-rs.

use arrow::array::{Array, ArrayRef, AsArray, BooleanArray};
use arrow::compute::kernels::comparison::contains as arrow_contains;
use arrow::datatypes::DataType;
use datafusion::common::{exec_err, Result, ScalarValue};
use datafusion::logical_expr::{
ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
};
use memchr::memmem::Finder;
use std::any::Any;
use std::sync::Arc;

/// Spark-optimized contains function.
///
/// Returns true if the first string argument contains the second string argument.
/// Optimized for the common case where the pattern is a scalar constant.
#[derive(Debug, PartialEq, Eq, Hash)]
pub struct SparkContains {
signature: Signature,
}

impl Default for SparkContains {
fn default() -> Self {
Self::new()
}
}

impl SparkContains {
pub fn new() -> Self {
Self {
signature: Signature::variadic_any(Volatility::Immutable),
}
}
}

impl ScalarUDFImpl for SparkContains {
fn as_any(&self) -> &dyn Any {
self
}

fn name(&self) -> &str {
"contains"
}

fn signature(&self) -> &Signature {
&self.signature
}

fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
Ok(DataType::Boolean)
}

fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
if args.args.len() != 2 {
return exec_err!("contains function requires exactly 2 arguments");
}
spark_contains(&args.args[0], &args.args[1])
}
}

/// Execute the contains function with optimized scalar pattern handling.
fn spark_contains(haystack: &ColumnarValue, needle: &ColumnarValue) -> Result<ColumnarValue> {
match (haystack, needle) {
// Case 1: Both are arrays - use arrow's contains directly
(ColumnarValue::Array(haystack_array), ColumnarValue::Array(needle_array)) => {
let result = arrow_contains(haystack_array, needle_array)?;
Ok(ColumnarValue::Array(Arc::new(result)))
}

// Case 2: Haystack is array, needle is scalar - OPTIMIZED PATH
// This is the common case in SQL like: WHERE col CONTAINS 'pattern'
(ColumnarValue::Array(haystack_array), ColumnarValue::Scalar(needle_scalar)) => {
let result = contains_with_scalar_pattern(haystack_array, needle_scalar)?;
Ok(ColumnarValue::Array(result))
}

// Case 3: Haystack is scalar, needle is array - less common
(ColumnarValue::Scalar(haystack_scalar), ColumnarValue::Array(needle_array)) => {
// Convert scalar to array and use arrow's contains
let haystack_array = haystack_scalar.to_array_of_size(needle_array.len())?;
let result = arrow_contains(&haystack_array, needle_array)?;
Ok(ColumnarValue::Array(Arc::new(result)))
}

// Case 4: Both are scalars - compute single result
(ColumnarValue::Scalar(haystack_scalar), ColumnarValue::Scalar(needle_scalar)) => {
let result = contains_scalar_scalar(haystack_scalar, needle_scalar)?;
Ok(ColumnarValue::Scalar(result))
}
}
}

/// Optimized contains for array haystack with scalar needle pattern.
/// Uses memchr's SIMD-optimized Finder for efficient repeated searches.
fn contains_with_scalar_pattern(
haystack_array: &ArrayRef,
needle_scalar: &ScalarValue,
) -> Result<ArrayRef> {
// Handle null needle
if needle_scalar.is_null() {
return Ok(Arc::new(BooleanArray::new_null(haystack_array.len())));
}

// Extract the needle string
let needle_str = match needle_scalar {
ScalarValue::Utf8(Some(s))
| ScalarValue::LargeUtf8(Some(s))
| ScalarValue::Utf8View(Some(s)) => s.as_str(),
_ => {
return exec_err!(
"contains function requires string type for needle, got {:?}",
needle_scalar.data_type()
)
}
};

// Create a reusable Finder for efficient SIMD-optimized searching
let finder = Finder::new(needle_str.as_bytes());

match haystack_array.data_type() {
DataType::Utf8 => {
let array = haystack_array.as_string::<i32>();
let result: BooleanArray = array
.iter()
.map(|opt_haystack| opt_haystack.map(|h| finder.find(h.as_bytes()).is_some()))
.collect();
Comment on lines +149 to +152
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

did you try using arrow's contains kernel here?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To use Arrow's kernel for scalar patterns, we need arrow_string::like::contains
which need arrow-string dependency

#2991 (comment)

Ok(Arc::new(result))
}
DataType::LargeUtf8 => {
let array = haystack_array.as_string::<i64>();
let result: BooleanArray = array
.iter()
.map(|opt_haystack| opt_haystack.map(|h| finder.find(h.as_bytes()).is_some()))
.collect();
Ok(Arc::new(result))
}
DataType::Utf8View => {
let array = haystack_array.as_string_view();
let result: BooleanArray = array
.iter()
.map(|opt_haystack| opt_haystack.map(|h| finder.find(h.as_bytes()).is_some()))
.collect();
Ok(Arc::new(result))
}
other => exec_err!(
"contains function requires string type for haystack, got {:?}",
other
),
}
}

/// Contains for two scalar values.
fn contains_scalar_scalar(
haystack_scalar: &ScalarValue,
needle_scalar: &ScalarValue,
) -> Result<ScalarValue> {
// Handle nulls
if haystack_scalar.is_null() || needle_scalar.is_null() {
return Ok(ScalarValue::Boolean(None));
}

let haystack_str = match haystack_scalar {
ScalarValue::Utf8(Some(s))
| ScalarValue::LargeUtf8(Some(s))
| ScalarValue::Utf8View(Some(s)) => s.as_str(),
_ => {
return exec_err!(
"contains function requires string type for haystack, got {:?}",
haystack_scalar.data_type()
)
}
};

let needle_str = match needle_scalar {
ScalarValue::Utf8(Some(s))
| ScalarValue::LargeUtf8(Some(s))
| ScalarValue::Utf8View(Some(s)) => s.as_str(),
_ => {
return exec_err!(
"contains function requires string type for needle, got {:?}",
needle_scalar.data_type()
)
}
};

Ok(ScalarValue::Boolean(Some(
haystack_str.contains(needle_str),
)))
}

#[cfg(test)]
mod tests {
use super::*;
use arrow::array::StringArray;

#[test]
fn test_contains_array_scalar() {
let haystack = Arc::new(StringArray::from(vec![
Some("hello world"),
Some("foo bar"),
Some("testing"),
None,
])) as ArrayRef;
let needle = ScalarValue::Utf8(Some("world".to_string()));

let result = contains_with_scalar_pattern(&haystack, &needle).unwrap();
let bool_array = result.as_any().downcast_ref::<BooleanArray>().unwrap();

assert!(bool_array.value(0)); // "hello world" contains "world"
assert!(!bool_array.value(1)); // "foo bar" does not contain "world"
assert!(!bool_array.value(2)); // "testing" does not contain "world"
assert!(bool_array.is_null(3)); // null input => null output
}

#[test]
fn test_contains_scalar_scalar() {
let haystack = ScalarValue::Utf8(Some("hello world".to_string()));
let needle = ScalarValue::Utf8(Some("world".to_string()));

let result = contains_scalar_scalar(&haystack, &needle).unwrap();
assert_eq!(result, ScalarValue::Boolean(Some(true)));

let needle_not_found = ScalarValue::Utf8(Some("xyz".to_string()));
let result = contains_scalar_scalar(&haystack, &needle_not_found).unwrap();
assert_eq!(result, ScalarValue::Boolean(Some(false)));
}

#[test]
fn test_contains_null_needle() {
let haystack = Arc::new(StringArray::from(vec![
Some("hello world"),
Some("foo bar"),
])) as ArrayRef;
let needle = ScalarValue::Utf8(None);

let result = contains_with_scalar_pattern(&haystack, &needle).unwrap();
let bool_array = result.as_any().downcast_ref::<BooleanArray>().unwrap();

// Null needle should produce null results
assert!(bool_array.is_null(0));
assert!(bool_array.is_null(1));
}

#[test]
fn test_contains_empty_needle() {
let haystack = Arc::new(StringArray::from(vec![Some("hello world"), Some("")])) as ArrayRef;
let needle = ScalarValue::Utf8(Some("".to_string()));

let result = contains_with_scalar_pattern(&haystack, &needle).unwrap();
let bool_array = result.as_any().downcast_ref::<BooleanArray>().unwrap();

// Empty string is contained in any string
assert!(bool_array.value(0));
assert!(bool_array.value(1));
}
}
2 changes: 2 additions & 0 deletions native/spark-expr/src/string_funcs/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,10 @@
// specific language governing permissions and limitations
// under the License.

mod contains;
mod string_space;
mod substring;

pub use contains::SparkContains;
pub use string_space::SparkStringSpace;
pub use substring::SubstringExpr;
20 changes: 20 additions & 0 deletions pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -404,6 +404,26 @@ under the License.
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-avro_${scala.binary.version}</artifactId>
<version>${spark.version}</version>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.apache.datafusion</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.arrow</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
Expand Down
5 changes: 5 additions & 0 deletions spark/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,11 @@ under the License.
<artifactId>spark-hadoop-cloud_${scala.binary.version}</artifactId>
<classifier>tests</classifier>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-avro_${scala.binary.version}</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
Expand Down
19 changes: 18 additions & 1 deletion spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
Original file line number Diff line number Diff line change
Expand Up @@ -1107,7 +1107,24 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {

// Filter rows that contains 'rose' in 'name' column
val queryContains = sql(s"select id from $table where contains (name, 'rose')")
checkAnswer(queryContains, Row(5) :: Nil)
checkSparkAnswerAndOperator(queryContains)

// Additional test cases for optimized contains implementation
// Test with empty pattern (should match all non-null rows)
val queryEmptyPattern = sql(s"select id from $table where contains (name, '')")
checkSparkAnswerAndOperator(queryEmptyPattern)

// Test with pattern not found
val queryNotFound = sql(s"select id from $table where contains (name, 'xyz')")
checkSparkAnswerAndOperator(queryNotFound)

// Test with pattern at start
val queryStart = sql(s"select id from $table where contains (name, 'James')")
checkSparkAnswerAndOperator(queryStart)

// Test with pattern at end
val queryEnd = sql(s"select id from $table where contains (name, 'Smith')")
checkSparkAnswerAndOperator(queryEnd)
}
}

Expand Down
Loading