refactor: introduce REGEXP_ENGINE_RUST/REGEXP_ENGINE_JAVA constants

Replace string literals "rust"/"java" used for the regexp engine selector
with named constants on CometConf. Tighten CometRLike.getSupportLevel so
it only reports Compatible(None) when the pattern is a Literal, matching
the actual constraint enforced by the convert path.

Author: andygrove

common/src/main/scala/org/apache/comet/CometConf.scala

@@ -383,19 +383,22 @@ object CometConf extends ShimCometConf {
       .booleanConf
       .createWithDefault(false)
 
+  val REGEXP_ENGINE_RUST = "rust"
+  val REGEXP_ENGINE_JAVA = "java"
+
   val COMET_REGEXP_ENGINE: ConfigEntry[String] =
     conf("spark.comet.exec.regexp.engine")
       .category(CATEGORY_EXEC)
       .doc(
         "Experimental. Selects the engine used to evaluate supported regular-expression " +
-          "expressions. `rust` uses the native DataFusion regexp engine. `java` routes through " +
-          "a JVM-side UDF (java.util.regex.Pattern) for Spark-compatible semantics, at the " +
-          "cost of JNI roundtrips per batch. Only RLike is routed today; additional " +
-          "expressions may opt in over time.")
+          s"expressions. `$REGEXP_ENGINE_RUST` uses the native DataFusion regexp engine. " +
+          s"`$REGEXP_ENGINE_JAVA` routes through a JVM-side UDF (java.util.regex.Pattern) for " +
+          "Spark-compatible semantics, at the cost of JNI roundtrips per batch. Only RLike " +
+          "is routed today; additional expressions may opt in over time.")
       .stringConf
       .transform(_.toLowerCase(Locale.ROOT))
-      .checkValues(Set("rust", "java"))
-      .createWithDefault("rust")
+      .checkValues(Set(REGEXP_ENGINE_RUST, REGEXP_ENGINE_JAVA))
+      .createWithDefault(REGEXP_ENGINE_RUST)
 
   val COMET_EXEC_SHUFFLE_WITH_HASH_PARTITIONING_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.native.shuffle.partitioning.hash.enabled")

spark/src/main/scala/org/apache/comet/serde/strings.scala

@@ -267,15 +267,18 @@ object CometRLike extends CometExpressionSerde[RLike] {
     "Uses Rust regexp engine, which has different behavior to Java regexp engine")
 
   override def getSupportLevel(expr: RLike): SupportLevel = {
-    if (CometConf.COMET_REGEXP_ENGINE.get() == "java") {
-      Compatible(None)
+    if (CometConf.COMET_REGEXP_ENGINE.get() == CometConf.REGEXP_ENGINE_JAVA) {
+      expr.right match {
+        case _: Literal => Compatible(None)
+        case _ => Unsupported(Some("Only scalar regexp patterns are supported"))
+      }
     } else {
       super.getSupportLevel(expr)
     }
   }
 
   override def convert(expr: RLike, inputs: Seq[Attribute], binding: Boolean): Option[Expr] = {
-    if (CometConf.COMET_REGEXP_ENGINE.get() == "java") {
+    if (CometConf.COMET_REGEXP_ENGINE.get() == CometConf.REGEXP_ENGINE_JAVA) {
       convertViaJvmUdf(expr, inputs, binding)
     } else {
       convertViaNativeRegex(expr, inputs, binding)

spark/src/test/scala/org/apache/comet/CometRegExpJvmSuite.scala

@@ -21,52 +21,157 @@ package org.apache.comet
 
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.CometTestBase
-import org.apache.spark.sql.comet.CometProjectExec
+import org.apache.spark.sql.comet.{CometFilterExec, CometProjectExec}
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
 
 class CometRegExpJvmSuite extends CometTestBase with AdaptiveSparkPlanHelper {
 
   override protected def sparkConf: SparkConf =
-    super.sparkConf.set("spark.comet.exec.regexp.engine", "java")
+    super.sparkConf.set(CometConf.COMET_REGEXP_ENGINE.key, CometConf.REGEXP_ENGINE_JAVA)
 
-  test("rlike: Java regex semantics, with flag on") {
+  // Patterns that the Rust regex crate cannot handle. Using one of these proves
+  // the JVM path was taken: if the pattern reached native, native would have
+  // rejected it and the operator would not be Comet.
+  private val backreference = "^(\\\\w)\\\\1$"
+  private val lookahead = "foo(?=bar)"
+  private val lookbehind = "(?<=foo)bar"
+  private val embeddedFlags = "(?i)foo"
+  private val namedGroup = "(?<digit>\\\\d)"
+
+  private def withSubjects(values: String*)(f: => Unit): Unit = {
     withTable("t") {
       sql("CREATE TABLE t (s STRING) USING parquet")
-      sql("INSERT INTO t VALUES ('abc123'), ('no digits'), (NULL), ('mixed_42_data')")
+      val rows = values
+        .map(v => if (v == null) "(NULL)" else s"('${v.replace("'", "''")}')")
+        .mkString(", ")
+      sql(s"INSERT INTO t VALUES $rows")
+      f
+    }
+  }
+
+  test("rlike: projection produces Java regex semantics with null handling") {
+    withSubjects("abc123", "no digits", null, "mixed_42_data") {
       val df = sql("SELECT s, s rlike '\\\\d+' AS m FROM t")
-      val rows = df.collect().map { r =>
-        val matched: Any = if (r.isNullAt(1)) null else r.getBoolean(1)
-        (Option(r.getString(0)), matched)
-      }
+      checkSparkAnswerAndOperator(df)
+    }
+  }
+
+  test("rlike: predicate filters rows using Java regex semantics") {
+    withSubjects("abc123", "no digits", null, "mixed_42_data") {
+      val df = sql("SELECT s FROM t WHERE s rlike '\\\\d+'")
+      checkSparkAnswerAndOperator(df)
+    }
+  }
+
+  test("rlike: backreference in projection (Java-only construct)") {
+    withSubjects("aa", "ab", "xyzzy", null) {
+      val df = sql(s"SELECT s, s rlike '$backreference' FROM t")
+      checkSparkAnswerAndOperator(df)
+      val plan = df.queryExecution.executedPlan
       assert(
-        rows.toSet === Set(
-          (Some("abc123"), true),
-          (Some("no digits"), false),
-          (None, null),
-          (Some("mixed_42_data"), true)))
+        collect(plan) { case p: CometProjectExec => p }.nonEmpty,
+        s"Expected CometProjectExec in:\n$plan")
     }
   }
 
-  test("rlike: plan contains CometProjectExec when flag is on") {
-    withTable("t") {
-      sql("CREATE TABLE t (s STRING) USING parquet")
-      sql("INSERT INTO t VALUES ('a1')")
-      val df = sql("SELECT s rlike '[a-z]\\\\d' FROM t")
-      df.collect()
-      val cometProjects = collect(df.queryExecution.executedPlan) { case p: CometProjectExec =>
-        p
-      }
+  test("rlike: backreference in predicate (Java-only construct)") {
+    withSubjects("aa", "ab", "xyzzy", null) {
+      val df = sql(s"SELECT s FROM t WHERE s rlike '$backreference'")
+      checkSparkAnswerAndOperator(df)
+      val plan = df.queryExecution.executedPlan
+      assert(
+        collect(plan) { case f: CometFilterExec => f }.nonEmpty,
+        s"Expected CometFilterExec in:\n$plan")
+    }
+  }
+
+  test("rlike: lookahead pattern (Java-only construct)") {
+    withSubjects("foobar", "foobaz", "barfoo", null) {
+      checkSparkAnswerAndOperator(sql(s"SELECT s, s rlike '$lookahead' FROM t"))
+      checkSparkAnswerAndOperator(sql(s"SELECT s FROM t WHERE s rlike '$lookahead'"))
+    }
+  }
+
+  test("rlike: lookbehind pattern (Java-only construct)") {
+    withSubjects("foobar", "barbar", "foofoo", null) {
+      checkSparkAnswerAndOperator(sql(s"SELECT s, s rlike '$lookbehind' FROM t"))
+    }
+  }
+
+  test("rlike: embedded case-insensitive flag (Java-only construct)") {
+    withSubjects("FOO", "foo", "fOO", "bar") {
+      checkSparkAnswerAndOperator(sql(s"SELECT s, s rlike '$embeddedFlags' FROM t"))
+    }
+  }
+
+  test("rlike: named groups (Java-only construct)") {
+    withSubjects("a1", "ab", "9z", null) {
+      checkSparkAnswerAndOperator(sql(s"SELECT s, s rlike '$namedGroup' FROM t"))
+    }
+  }
+
+  test("rlike: empty pattern matches every non-null row") {
+    withSubjects("abc", "", null) {
+      checkSparkAnswerAndOperator(sql("SELECT s, s rlike '' FROM t"))
+    }
+  }
+
+  test("rlike: empty subject string is handled correctly") {
+    withSubjects("", "x", null) {
+      checkSparkAnswerAndOperator(sql("SELECT s, s rlike '^$' FROM t"))
+    }
+  }
+
+  test("rlike: all-null subject column produces all-null result") {
+    withSubjects(null, null, null) {
+      checkSparkAnswerAndOperator(sql("SELECT s rlike '\\\\d+' FROM t"))
+    }
+  }
+
+  test("rlike: null literal pattern falls back to Spark") {
+    withSubjects("a", "b", null) {
+      // Convert path rejects Literal(null) pattern; query must still produce
+      // Spark-compatible all-null output via the Spark fallback.
+      checkSparkAnswer(sql("SELECT s rlike CAST(NULL AS STRING) FROM t"))
+    }
+  }
+
+  test("rlike: invalid pattern falls back to Spark") {
+    withSubjects("a") {
+      // Convert path catches PatternSyntaxException at planning time; Spark's
+      // own RLike runs and throws its native error.
+      val ex = intercept[Throwable](sql("SELECT s rlike '[' FROM t").collect())
       assert(
-        cometProjects.nonEmpty,
-        s"Expected at least one CometProjectExec in:\n${df.queryExecution.executedPlan}")
+        ex.getMessage.toLowerCase.contains("regex") ||
+          ex.getMessage.contains("PatternSyntax") ||
+          ex.getMessage.contains("Unclosed"),
+        s"Unexpected error: ${ex.getMessage}")
+    }
+  }
+
+  test("rlike: combines with filter, projection, and aggregate") {
+    withTable("t") {
+      sql("CREATE TABLE t (s STRING, k INT) USING parquet")
+      sql("""INSERT INTO t VALUES
+            |  ('aa', 1), ('ab', 1), ('aa', 2), ('xyzzy', 2), ('aa', 3), (NULL, 3)""".stripMargin)
+      val df = sql(s"""SELECT k, COUNT(*) AS c
+           |FROM t
+           |WHERE s rlike '$backreference'
+           |GROUP BY k
+           |ORDER BY k""".stripMargin)
+      checkSparkAnswerAndOperator(df)
     }
   }
 
-  test("rlike: result matches Spark for arbitrary patterns") {
+  test("rlike: many rows spanning multiple batches") {
     withTable("t") {
       sql("CREATE TABLE t (s STRING) USING parquet")
-      sql("INSERT INTO t VALUES ('123'), ('abc'), ('mix3d')")
-      checkSparkAnswer(sql("SELECT s, s rlike '\\\\d+' FROM t"))
+      val values = (0 until 5000)
+        .map(i => if (i % 7 == 0) "(NULL)" else s"('row_${i}_aa')")
+        .mkString(", ")
+      sql(s"INSERT INTO t VALUES $values")
+      checkSparkAnswerAndOperator(sql(s"SELECT s, s rlike '$backreference' FROM t"))
+      checkSparkAnswerAndOperator(sql(s"SELECT s FROM t WHERE s rlike '$backreference'"))
     }
   }
 }

spark/src/test/scala/org/apache/spark/sql/benchmark/CometRegExpBenchmark.scala

@@ -110,7 +110,8 @@ object CometRegExpBenchmark extends CometBenchmarkBase {
     }
 
     benchmark.addCase("Comet (Exec, JVM regex)") { _ =>
-      val configs = baseExec ++ Map(CometConf.COMET_REGEXP_ENGINE.key -> "java")
+      val configs =
+        baseExec ++ Map(CometConf.COMET_REGEXP_ENGINE.key -> CometConf.REGEXP_ENGINE_JAVA)
       withSQLConf(configs.toSeq: _*) {
         spark.sql(query).noop()
       }