apache · andygrove · Aug 3, 2024 · Aug 3, 2024 · Aug 3, 2024
diff --git a/native/Cargo.lock b/native/Cargo.lock
diff --git a/native/Cargo.toml b/native/Cargo.toml
@@ -53,6  53,7 @@ chrono-tz = { version = "0.8" }
 num = "0.4"
 rand = "0.8"
 regex = "1.9.6"
 regex-syntax = "0.8.4"
 thiserror = "1"
 
 [profile.release]

diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs
@@ -61,6  61,7 @@ use jni::{
 use tokio::runtime::Runtime;
 
 use crate::execution::operators::ScanExec;
 use datafusion_comet_spark_expr::is_regexp_supported;
 use log::info;
 
 /// Comet native execution context. Kept alive across JNI calls.
@@ -93,7  94,7 @@ struct ExecutionContext {
 
 /// Accept serialized query plan and return the address of the native query plan.
 /// # Safety
-/// This function is inheritly unsafe since it deals with raw pointers passed from JNI.
 /// This function is inherently unsafe since it deals with raw pointers passed from JNI.
 #[no_mangle]
 pub unsafe extern "system" fn Java_org_apache_comet_Native_createPlan(
     e: JNIEnv,
@@ -300,7  301,7 @@ fn pull_input_batches(exec_context: &mut ExecutionContext) -> Result<(), CometEr
 /// Accept serialized query plan and the addresses of Arrow Arrays from Spark,
 /// then execute the query. Return addresses of arrow vector.
 /// # Safety
-/// This function is inheritly unsafe since it deals with raw pointers passed from JNI.
 /// This function is inherently unsafe since it deals with raw pointers passed from JNI.
 #[no_mangle]
 pub unsafe extern "system" fn Java_org_apache_comet_Native_executePlan(
     e: JNIEnv,
@@ -450,7  451,7 @@ fn get_execution_context<'a>(id: i64) -> &'a mut ExecutionContext {
 
 /// Used by Comet shuffle external sorter to write sorted records to disk.
 /// # Safety
-/// This function is inheritly unsafe since it deals with raw pointers passed from JNI.
 /// This function is inherently unsafe since it deals with raw pointers passed from JNI.
 #[no_mangle]
 pub unsafe extern "system" fn Java_org_apache_comet_Native_writeSortedFileNative(
     e: JNIEnv,
@@ -535,3  536,19 @@ pub extern "system" fn Java_org_apache_comet_Native_sortRowPartitionsNative(
         Ok(())
     })
 }
 
 /// Used by QueryPlanSerde to determine if a regular expression is supported natively
 /// # Safety
 /// This function is inherently unsafe since it deals with raw pointers passed from JNI.
 #[no_mangle]
 pub unsafe extern "system" fn Java_org_apache_comet_Native_isRegexpPatternSupported(
     e: JNIEnv,
     _class: JClass,
     pattern: jstring,
 ) -> jboolean {
     try_unwrap_or_throw(&e, |mut env| {
         let pattern: String = env.get_string(&JString::from_raw(pattern)).unwrap().into();
         // if we hit an error parsing the regexp then just report it as unsupported
         Ok(is_regexp_supported(&pattern).ok().unwrap_or(false) as jboolean)
     })
 }
diff --git a/native/spark-expr/Cargo.toml b/native/spark-expr/Cargo.toml
@@ -40,6  40,7 @@ datafusion-physical-plan = { workspace = true }
 chrono-tz = { workspace = true }
 num = { workspace = true }
 regex = { workspace = true }
 regex-syntax = { workspace = true }
 thiserror = { workspace = true }
 unicode-segmentation = "1.11.0"
 
@@ -49,6  50,7 @@ criterion = "0.5.1"
 rand = { workspace = true}
 twox-hash = "1.6.3"
 
 
 [lib]
 name = "datafusion_comet_spark_expr"
 path = "src/lib.rs"

diff --git a/native/spark-expr/src/lib.rs b/native/spark-expr/src/lib.rs
@@ -21,6  21,7 @@ mod if_expr;
 
 mod kernels;
 mod regexp;
 pub use regexp::is_regexp_supported;
 pub mod scalar_funcs;
 pub mod spark_hash;
 mod structs;

diff --git a/native/spark-expr/src/regexp.rs b/native/spark-expr/src/regexp.rs
@@ -22,10  22,12 @@ use arrow_array::builder::BooleanBuilder;
 use arrow_array::types::Int32Type;
 use arrow_array::{Array, BooleanArray, DictionaryArray, RecordBatch, StringArray};
 use arrow_schema::{DataType, Schema};
-use datafusion_common::{internal_err, Result};
 use datafusion_common::{internal_err, DataFusionError, Result};
 use datafusion_expr::ColumnarValue;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 use regex::Regex;
 use regex_syntax::hir::{Class, Hir, HirKind};
 use regex_syntax::Parser;
 use std::any::Any;
 use std::fmt::{Display, Formatter};
 use std::hash::{Hash, Hasher};
@@ -47,6  49,27 @@ pub struct RLike {
     pattern: Regex,
 }
 
 /// Determine if a regex pattern is guaranteed to produce the same values as
 /// Java's regular expression engine
 pub fn is_regexp_supported(pattern: &str) -> Result<bool> {
     let ast = Parser::new()
         .parse(pattern)
         .map_err(|e| DataFusionError::Execution(format!("Error parsing regex pattern: {}", e)))?;
     Ok(is_compat(&ast))
 }
 
 fn is_compat(ast: &Hir) -> bool {
     match ast.kind() {
         // character class such as `[a-z]` or `[^aeiou]`
         HirKind::Class(Class::Unicode(c)) => c.is_ascii(),
         // repetition quantifier such as ` `, `*`, `?`, `{1,3}`
         HirKind::Repetition(r) => is_compat(r.sub.as_ref()),
         // series of expressions such as `[A-Z][a-z]`
         HirKind::Concat(items) => items.iter().all(is_compat),
         _ => false,
     }
 }
 
 impl Hash for RLike {
     fn hash<H: Hasher>(&self, state: &mut H) {
         state.write(self.pattern_str.as_bytes());
@@ -168,3  191,25 @@ impl PhysicalExpr for RLike {
         self.hash(&mut s);
     }
 }
 
 #[cfg(test)]
 mod test {
     use crate::regexp::is_regexp_supported;
     use datafusion_common::Result;
 
     #[test]
     fn parse_supported_regex() -> Result<()> {
         assert!(is_regexp_supported("[a-z]")?);
         assert!(is_regexp_supported("[a-z] ")?);
         assert!(is_regexp_supported("[a-z]*")?);
         assert!(is_regexp_supported("[a-z]?")?);
         Ok(())
     }
 
     #[test]
     fn parse_unsupported_regex() -> Result<()> {
         assert!(!is_regexp_supported("abc$")?);
         assert!(!is_regexp_supported("^abc")?);
         Ok(())
     }
 }
diff --git a/spark/src/main/scala/org/apache/comet/Native.scala b/spark/src/main/scala/org/apache/comet/Native.scala
@@ -121,4  121,15 @@ class Native extends NativeBase {
    *   the size of the array.
    */
   @native def sortRowPartitionsNative(addr: Long, size: Long): Unit
 
   /**
    * Determine if a regular expression pattern is supported natively and guaranteed to produce the
    * same result as Spark.
    *
    * @param pattern
    *   The regular expression pattern
    * @return
    *   true if supported
    */
   @native def isRegexpPatternSupported(pattern: String): Boolean
 }
diff --git a/spark/src/main/scala/org/apache/comet/expressions/RegExp.scala b/spark/src/main/scala/org/apache/comet/expressions/RegExp.scala
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -42,9  42,9 @@ import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.UTF8String
 
-import org.apache.comet.CometConf
 import org.apache.comet.{CometConf, Native}
 import org.apache.comet.CometSparkSessionExtensions.{isCometOperatorEnabled, isCometScan, isSpark34Plus, withInfo}
-import org.apache.comet.expressions.{CometCast, CometEvalMode, Compatible, Incompatible, RegExp, Unsupported}
 import org.apache.comet.expressions.{CometCast, CometEvalMode, Compatible, Incompatible, Unsupported}
 import org.apache.comet.serde.ExprOuterClass.{AggExpr, DataType => ProtoDataType, Expr, ScalarFunc}
 import org.apache.comet.serde.ExprOuterClass.DataType.{DataTypeInfo, DecimalInfo, ListInfo, MapInfo, StructInfo}
 import org.apache.comet.serde.OperatorOuterClass.{AggregateMode => CometAggregateMode, BuildSide, JoinType, Operator}
@@ -55,6  55,9 @@ import org.apache.comet.shims.ShimQueryPlanSerde
  * An utility object for query plan and expression serialization.
  */
 object QueryPlanSerde extends Logging with ShimQueryPlanSerde with CometExprShim {
 
   private val native = new Native()
 
   def emitWarning(reason: String): Unit = {
     logWarning(s"Comet native execution is disabled due to: $reason")
   }
@@ -1237,7  1240,7 @@ object QueryPlanSerde extends Logging with ShimQueryPlanSerde with CometExprShim
           // we currently only support scalar regex patterns
           right match {
             case Literal(pattern, DataTypes.StringType) =>
-              if (!RegExp.isSupportedPattern(pattern.toString) &&
               if (!native.isRegexpPatternSupported(pattern.toString) &&
                 !CometConf.COMET_REGEXP_ALLOW_INCOMPATIBLE.get()) {
                 withInfo(
                   expr,

diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
@@ -625,11  625,9 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
         // add repetitive data to trigger dictionary encoding
         Range(0, 100).map(_ => "John Smith")
       withParquetFile(data.zipWithIndex, withDictionary) { file =>
-        withSQLConf(CometConf.COMET_REGEXP_ALLOW_INCOMPATIBLE.key -> "true") {
-          spark.read.parquet(file).createOrReplaceTempView(table)
-          val query = sql(s"select _2 as id, _1 rlike 'R[a-z] s [Rr]ose' from $table")
-          checkSparkAnswerAndOperator(query)
-        }
         spark.read.parquet(file).createOrReplaceTempView(table)
         val query = sql(s"select _2 as id, _1 rlike '[M-R] [a-z] ' from $table")
         checkSparkAnswerAndOperator(query)
       }
     }
   }