diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml index 52e4af61117..03e37d8a6cf 100644 --- a/.github/workflows/pr_build_linux.yml +++ b/.github/workflows/pr_build_linux.yml @@ -400,6 +400,8 @@ jobs: org.apache.comet.expressions.conditional.CometCoalesceSuite org.apache.comet.expressions.conditional.CometCaseWhenSuite org.apache.comet.CometRegExpJvmSuite + org.apache.comet.CometRegexParitySuite + org.apache.comet.expressions.CometRegexSuite org.apache.comet.CometCodegenSuite org.apache.comet.CometCoverageStatsSuite org.apache.comet.CometCodegenSourceSuite diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml index 5095f8b5493..ea4d3efbd0e 100644 --- a/.github/workflows/pr_build_macos.yml +++ b/.github/workflows/pr_build_macos.yml @@ -216,6 +216,8 @@ jobs: org.apache.comet.expressions.conditional.CometCoalesceSuite org.apache.comet.expressions.conditional.CometCaseWhenSuite org.apache.comet.CometRegExpJvmSuite + org.apache.comet.CometRegexParitySuite + org.apache.comet.expressions.CometRegexSuite org.apache.comet.CometCodegenSuite org.apache.comet.CometCoverageStatsSuite org.apache.comet.CometCodegenSourceSuite diff --git a/docs/source/contributor-guide/expression-audits/predicate_funcs.md b/docs/source/contributor-guide/expression-audits/predicate_funcs.md index 57a55262ca8..13968021262 100644 --- a/docs/source/contributor-guide/expression-audits/predicate_funcs.md +++ b/docs/source/contributor-guide/expression-audits/predicate_funcs.md @@ -146,6 +146,6 @@ ## rlike -- See `string_funcs / regexp_replace` and the `CometRLike` notes (audited in PR [#4461](https://github.com/apache/datafusion-comet/pull/4461)). By default `CometRLike` routes through the JVM codegen dispatcher so Spark's own `Pattern` engine runs inside the Comet pipeline. The native path uses the Rust `regex` crate, which differs from Java's `Pattern` engine, so it is opt-in via `spark.comet.expression.RLike.allowIncompatible=true` and only applies when the pattern is a string literal. +- See `string_funcs / regexp_replace` and the `CometRLike` notes (audited in PR [#4461](https://github.com/apache/datafusion-comet/pull/4461)). `CometRLike` consults a plan-time whitelist (`CometRegex`) for non-null `UTF8_BINARY` literal patterns. Analyzer-admitted patterns (ASCII literals, simple character classes, greedy quantifiers, groups, alternation) run natively by default. Every other literal stays on the JVM codegen dispatcher unless `spark.comet.expression.RLike.allowIncompatible=true` forces the Rust path. Non-literal and NULL patterns, and Spark 4 non-default collations, always stay on the dispatcher. Anchors (`^`, `$`) and `.` / `\d` / `\w` / `\s` are out of subset because Java and Rust disagree on line terminators and Unicode character classes. Rust class set operators (`&&` / `~~` / `--`), unescaped class delimiters used as atoms or range endpoints, and counted, aggregate, or nested patterns over a conservative compile-size budget are also out of subset so they cannot silently change results, suppress Spark errors, or fail native plan construction. [Spark Expression Support]: ../../user-guide/latest/expressions.md diff --git a/docs/source/user-guide/latest/compatibility/regex.md b/docs/source/user-guide/latest/compatibility/regex.md index ffbab557d7b..02b8143ae71 100644 --- a/docs/source/user-guide/latest/compatibility/regex.md +++ b/docs/source/user-guide/latest/compatibility/regex.md @@ -22,16 +22,18 @@ under the License. Comet evaluates Spark regular-expression expressions (`rlike`, `regexp_replace`, `split`, `regexp_extract`, `regexp_extract_all`, `regexp_instr`) two ways: -- **Codegen dispatcher (default)** — Spark's own `doGenCode` for the expression runs inside Comet's +- **Codegen dispatcher** — Spark's own `doGenCode` for the expression runs inside Comet's Arrow-direct codegen dispatcher (the same dispatcher used by Comet's `ScalaUDF` codegen path). This is 100% compatible with Spark, at the cost of one JNI round-trip per batch. It is enabled by default (`spark.comet.exec.scalaUDF.codegen.enabled=true`); if the dispatcher is disabled, regex - expressions fall back to Spark. + expressions fall back to Spark. This is the default for every regex expression except an + in-subset `rlike` literal (see below). - **Native (rust) engine** — the Rust [`regex`] crate, run natively with no JNI overhead. It is - faster but has different semantics from Java regex (see below), so it is **opt-in per expression** - via that expression's `allowIncompatible` flag. `rlike`, `regexp_replace`, `split`, - `regexp_extract`, and `regexp_extract_all` have a native implementation; `regexp_instr` does not - and always runs through the codegen dispatcher. + faster but has different semantics from Java regex (see below). For `rlike`, a plan-time + analyzer admits a conservative subset of `UTF8_BINARY` literal patterns and runs those natively + **by default**. Every other `rlike` pattern, and every other regex expression, still requires + that expression's `allowIncompatible` flag. `regexp_instr` has no native implementation and + always runs through the codegen dispatcher. | SQL | Native (rust) opt-in config | | -------------------- | ----------------------------------------------------------- | @@ -41,9 +43,13 @@ Comet evaluates Spark regular-expression expressions (`rlike`, `regexp_replace`, | `regexp_extract_all` | `spark.comet.expression.RegExpExtractAll.allowIncompatible` | | `split` | `spark.comet.expression.StringSplit.allowIncompatible` | -When the native path is opted in but a case has no native implementation (for example a non-scalar -`rlike` pattern, `regexp_replace` with a non-1 offset, or `regexp_extract` with a non-literal -pattern or idx), Comet routes that case through the codegen dispatcher. +`spark.comet.expression.RLike.allowIncompatible` only forces the native Rust path for literal +patterns the analyzer cannot prove equivalent to Java regex. It is not needed for in-subset +literals, and it does not apply to non-literal or NULL patterns. + +When the native path is selected but a case has no native implementation (for example a +non-scalar `rlike` pattern, `regexp_replace` with a non-1 offset, or `regexp_extract` with a +non-literal pattern or idx), Comet routes that case through the codegen dispatcher. ## Disabling Comet for individual regex expressions @@ -63,19 +69,22 @@ the engine selector: ## Choosing an engine -| | Rust engine | Codegen dispatcher (default) | +| | Rust engine | Codegen dispatcher | | -------------------- | ------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------- | -| **Compatibility** | Differs from Java regex (see below) | 100% compatible with Spark | +| **Compatibility** | In-subset `rlike` literals match Java regex; other patterns may differ (see below) | 100% compatible with Spark | | **Feature coverage** | `rlike`, `regexp_replace`, `split`, `regexp_extract`, `regexp_extract_all` natively; `regexp_instr` via fallthrough | All regexp expressions (`rlike`, `regexp_extract`, `regexp_extract_all`, `regexp_instr`, `regexp_replace`, `split`) | | **Performance** | Fully native, no JNI overhead | One JNI round-trip per batch (Arrow vectors stay columnar) | | **Pattern support** | Linear-time subset only | All Java regex features (backreferences, lookaround, etc.) | +| **`rlike` default** | Used automatically for analyzer-admitted `UTF8_BINARY` literals | Used for every other `rlike` pattern | -The **Rust engine** is faster but cannot match Java regex semantics for every pattern. Opting in per -expression (for example `spark.comet.expression.RLike.allowIncompatible=true`) declares acceptance -of those differences. +The **Rust engine** is faster but cannot match Java regex semantics for every pattern. For `rlike`, +Comet therefore runs only the analyzer-admitted subset natively by default. Setting +`spark.comet.expression.RLike.allowIncompatible=true` forces the Rust path for other literal +patterns and declares acceptance of any remaining differences. The other regex expressions still +require their own `allowIncompatible` flag. -The **codegen dispatcher** is the default and is enabled by `spark.comet.exec.scalaUDF.codegen.enabled`, -so it can be disabled globally to fall back to Spark for the regex family. +The **codegen dispatcher** is enabled by `spark.comet.exec.scalaUDF.codegen.enabled`, so it can be +disabled globally to fall back to Spark for out-of-subset regex expressions. ## Why the engines differ @@ -131,13 +140,20 @@ Even where both engines accept a construct, the matching behavior is not always ## When the Rust engine is safe -For most ASCII-only, non-anchored patterns that use only literal characters, simple character classes, and -ordinary quantifiers, the two engines produce the same results. If you are confident your patterns fit this -shape and want to avoid the JNI overhead of the Java engine, switching to the Rust engine with -`allowIncompatible=true` is generally safe. +Comet's plan-time analyzer admits a conservative whitelist of `rlike` literals and runs those on +the Rust engine by default: printable ASCII literals, simple ASCII character classes, greedy +quantifiers (`*`, `+`, `?`, `{n}`, `{n,}`, `{n,m}`), capturing and non-capturing groups, and +alternation. Anchors (`^`, `$`), `.`, `\d` / `\w` / `\s`, inline flags, lookaround, Rust-only +class set operators (`&&`, `~~`, `--`), unescaped class delimiters used as atoms or range +endpoints, and counted, aggregate, or nested patterns that exceed a conservative compile-size +budget stay on the Java engine. Any unrecognized construct also stays on the Java engine. + +For `regexp_replace`, `split`, `regexp_extract`, and `regexp_extract_all`, the native path is +still opt-in via `allowIncompatible`. If you are confident those patterns fit the same ASCII, +non-anchored shape, opting in is generally safe. -For anything that uses backreferences, lookaround, or relies on Java's specific Unicode or line-handling -defaults, use the Java engine. +For anything that uses backreferences, lookaround, or relies on Java's specific Unicode or +line-handling defaults, use the Java engine. [`java.util.regex`]: https://docs.oracle.com/javase/8/docs/api/java/util/regex/Pattern.html [`regex`]: https://docs.rs/regex/latest/regex/ diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 192eae807eb..e6a09a0d691 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -541,9 +541,9 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | `like` | ✅ | Hybrid | | | `not` | ✅ | Native | | | `or` | ✅ | Native | | -| `regexp` | ✅ | Hybrid | Falls back by default; opt-in via allowIncompatible ([details](compatibility/regex.md)) | -| `regexp_like` | ✅ | Hybrid | Falls back by default; opt-in via allowIncompatible ([details](compatibility/regex.md)) | -| `rlike` | ✅ | Hybrid | Falls back by default; opt-in via allowIncompatible ([details](compatibility/regex.md)) | +| `regexp` | ✅ | Hybrid | In-subset literals run natively; others fall back by default ([details](compatibility/regex.md)) | +| `regexp_like` | ✅ | Hybrid | In-subset literals run natively; others fall back by default ([details](compatibility/regex.md)) | +| `rlike` | ✅ | Hybrid | In-subset literals run natively; others fall back by default ([details](compatibility/regex.md)) | --- diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometRegex.scala b/spark/src/main/scala/org/apache/comet/expressions/CometRegex.scala new file mode 100644 index 00000000000..1fa7149f937 --- /dev/null +++ b/spark/src/main/scala/org/apache/comet/expressions/CometRegex.scala @@ -0,0 +1,360 @@ +/* + * 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. + */ + +package org.apache.comet.expressions + +import org.apache.comet.serde.{Compatible, Incompatible, SupportLevel} + +/** + * Regex flavor for [[CometRegex]]. The first version only implements [[RegexFlavor.RLike]]; later + * flavors (for example `regexp_replace` / `split`) can add extra reject rules such as empty-match + * divergence without changing the scanner's whitelist core. + */ +sealed trait RegexFlavor + +object RegexFlavor { + case object RLike extends RegexFlavor +} + +/** + * Plan-time whitelist analyzer for literal Java regex patterns. A pattern is [[Compatible]] only + * when every construct is one the analyzer positively recognizes as equivalent on Spark's + * `java.util.regex` engine and Comet's Rust `regex` crate. Anything unrecognized is + * [[Incompatible]]: the safe direction, so a missed construct never silently takes the native + * path. + * + * This is a recursive-descent scan, not a search for forbidden substrings. `[(?=]` is a character + * class of literals, not a lookahead; `\\d` is a literal backslash plus `d`, not a digit class. + */ +object CometRegex { + + def supportLevel(pattern: String, flavor: RegexFlavor = RegexFlavor.RLike): SupportLevel = { + flavor match { + case RegexFlavor.RLike => + val scanner = new Scanner(pattern) + if (scanner.parseExpr().isDefined && !scanner.remaining) { + Compatible() + } else { + Incompatible(None) + } + } + } + + private val MetaEscapes: Set[Char] = + Set('.', '*', '+', '?', '(', ')', '[', ']', '{', '}', '|', '^', '$', '\\') + + // Conservative compile-size gates. Rust `regex` rejects large counted + // expansions and deep nesting; a single `{n}` cap is not enough because + // nested repetition multiplies. Stay well below crate defaults (nest 250). + private val MaxGroupDepth = 32 + private val MaxCountedBound = 256 + private val MaxExpansion = 4096L + + private class Scanner(pattern: String) { + private var i = 0 + private var groupDepth = 0 + + def remaining: Boolean = i < pattern.length + + private def peek: Char = pattern.charAt(i) + + private def peekOffset(n: Int): Option[Char] = { + val idx = i + n + if (idx < pattern.length) Some(pattern.charAt(idx)) else None + } + + private def consume(): Char = { + val c = peek + i += 1 + c + } + + private def startsWith(s: String): Boolean = pattern.startsWith(s, i) + + // Returns a conservative compiled-size estimate, or None if the construct + // is unrecognized or exceeds the compile budget. + def parseExpr(): Option[Long] = { + val first = parseTerm() match { + case Some(s) => s + case None => return None + } + var total = first + while (remaining && peek == '|') { + consume() + parseTerm() match { + case Some(s) => + addWithinBudget(total, s) match { + case Some(next) => total = next + case None => return None + } + case None => return None + } + } + Some(total) + } + + private def parseTerm(): Option[Long] = { + var total = 0L + var any = false + while (remaining && peek != '|' && peek != ')') { + parseFactor() match { + case Some(s) => + addWithinBudget(total, s) match { + case Some(next) => + total = next + any = true + case None => return None + } + case None => + return None + } + } + Some(if (any) total else 1L) + } + + private def parseFactor(): Option[Long] = { + val atomSize = parseAtom() match { + case Some(s) => s + case None => return None + } + parseOptionalQuantifier(atomSize) + } + + private def parseAtom(): Option[Long] = { + if (!remaining) { + return None + } + peek match { + case '\\' => + if (parseEscape(inClass = false).isDefined) Some(1L) else None + case '[' => + if (parseClass()) Some(1L) else None + case '(' => parseGroup() + case '.' | '^' | '$' | '*' | '+' | '?' | '{' | '}' | ')' | ']' | '|' => + None + case c if isPrintableAscii(c) => + consume() + Some(1L) + case _ => None + } + } + + private def parseGroup(): Option[Long] = { + consume() // '(' + if (!remaining) { + return None + } + if (startsWith("?:")) { + i += 2 + } else if (peek == '?') { + // lookaround, flags, named groups, atomic groups, comments, ... + return None + } + if (groupDepth >= MaxGroupDepth) { + return None + } + groupDepth += 1 + val inner = parseExpr() + val closed = remaining && consume() == ')' + groupDepth -= 1 + if (closed) inner else None + } + + private def parseOptionalQuantifier(atomSize: Long): Option[Long] = { + if (!remaining) { + return Some(atomSize) + } + peek match { + case '*' | '+' | '?' => + consume() + if (remaining && (peek == '+' || peek == '?')) { + // possessive or lazy + None + } else { + Some(atomSize) + } + case '{' => parseCountedQuantifier(atomSize) + case _ => Some(atomSize) + } + } + + private def parseCountedQuantifier(atomSize: Long): Option[Long] = { + consume() // '{' + val n = parseNonNegInt() match { + case Some(v) => v + case None => return None + } + if (n > MaxCountedBound) { + return None + } + if (!remaining) { + return None + } + val bound = peek match { + case '}' => + consume() + if (isLazyOrPossessiveSuffix) { + return None + } + n + case ',' => + consume() + if (!remaining) { + return None + } + if (peek == '}') { + consume() + if (isLazyOrPossessiveSuffix) { + return None + } + math.max(1, n) + } else { + val m = parseNonNegInt() match { + case Some(v) => v + case None => return None + } + if (m < n || m > MaxCountedBound) { + return None + } + if (!(remaining && consume() == '}' && !isLazyOrPossessiveSuffix)) { + return None + } + m + } + case _ => + return None + } + multiplyWithinBudget(atomSize, bound.toLong) + } + + private def addWithinBudget(a: Long, b: Long): Option[Long] = { + val total = saturatingAdd(a, b) + if (total > MaxExpansion) None else Some(total) + } + + private def multiplyWithinBudget(a: Long, b: Long): Option[Long] = { + // A zero-count repetition still contributes syntax and compile work. + // Keep every factor visible to aggregate term/alternation accounting. + val total = math.max(1L, saturatingMul(a, b)) + if (total > MaxExpansion) None else Some(total) + } + + private def saturatingAdd(a: Long, b: Long): Long = { + val s = a + b + if (s < 0) Long.MaxValue else s + } + + private def saturatingMul(a: Long, b: Long): Long = { + if (a != 0 && b > Long.MaxValue / a) { + Long.MaxValue + } else { + a * b + } + } + + private def isLazyOrPossessiveSuffix: Boolean = + remaining && (peek == '+' || peek == '?') + + private def parseNonNegInt(): Option[Int] = { + if (!remaining || !isAsciiDigit(peek)) { + return None + } + var v = 0L + while (remaining && isAsciiDigit(peek)) { + v = v * 10 + (consume() - '0') + if (v > Int.MaxValue) { + return None + } + } + Some(v.toInt) + } + + private def parseClass(): Boolean = { + consume() // '[' + if (remaining && peek == '^') { + consume() + } + var contentStarted = false + var lastAtom: Option[Char] = None + while (remaining && !(peek == ']' && contentStarted)) { + // Rust class set ops (&& / ~~ / --) are not Java literals. Nested + // classes and unescaped class delimiters as atoms are also out of subset. + if (startsWith("&&") || startsWith("~~") || startsWith("--") || peek == '[') { + return false + } + val ranging = lastAtom.isDefined && peek == '-' && peekOffset(1).exists(_ != ']') + if (ranging) { + consume() // '-' + parseClassAtom() match { + case Some(end) if end >= lastAtom.get => + lastAtom = None + case _ => + return false + } + } else { + parseClassAtom() match { + case Some(c) => + lastAtom = Some(c) + contentStarted = true + case None => + return false + } + } + } + remaining && consume() == ']' + } + + private def parseClassAtom(): Option[Char] = { + if (!remaining) { + return None + } + // Unescaped `]` is only the class closer, never a range endpoint. An + // unescaped `[` starts a nested class in Java, so it cannot be a literal + // range endpoint even though Rust accepts it as one. + if (peek == ']' || peek == '[') { + return None + } + if (peek == '\\') { + parseEscape(inClass = true) + } else if (isPrintableAscii(peek)) { + Some(consume()) + } else { + None + } + } + + private def parseEscape(inClass: Boolean): Option[Char] = { + consume() // '\' + if (!remaining) { + return None + } + val c = peek + val allowed = MetaEscapes.contains(c) || (inClass && c == '-') + if (allowed) { + Some(consume()) + } else { + None + } + } + + private def isPrintableAscii(c: Char): Boolean = c >= 0x20 && c <= 0x7e + + private def isAsciiDigit(c: Char): Boolean = c >= '0' && c <= '9' + } +} diff --git a/spark/src/main/scala/org/apache/comet/serde/strings.scala b/spark/src/main/scala/org/apache/comet/serde/strings.scala index ebf45089882..dbc66342000 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -21,8 +21,10 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Attribute, Base64, BitLength, Cast, Concat, ConcatWs, Contains, Elt, Empty2Null, EndsWith, Expression, FindInSet, FormatNumber, FormatString, GetJsonObject, InitCap, Left, Length, Levenshtein, Like, Literal, Lower, Mask, OctetLength, Overlay, RegExpExtract, RegExpExtractAll, RegExpInStr, RegExpReplace, Right, RLike, SoundEx, StartsWith, StringLocate, StringLPad, StringRepeat, StringReplace, StringRPad, StringSplit, StringTranslate, Substring, SubstringIndex, ToCharacter, ToNumber, TryToNumber, UnBase64, Upper} import org.apache.spark.sql.types.{BinaryType, DataTypes, IntegerType, LongType, StringType} +import org.apache.spark.unsafe.types.UTF8String import org.apache.comet.CometConf +import org.apache.comet.expressions.{CometRegex, RegexFlavor} import org.apache.comet.serde.ExprOuterClass.Expr import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType} import org.apache.comet.shims.CometTypeShim @@ -359,33 +361,62 @@ object CometEndsWith with CollationAwareBinaryPredicate[EndsWith] /** - * `rlike` runs Spark's own implementation through the codegen dispatcher by default, for - * byte-exact results. The native (rust) regexp engine is faster but has different semantics from - * Java regexp, so it is opt-in via `spark.comet.expression.RLike.allowIncompatible`; any case it - * does not cover (a non-scalar pattern) falls through to the codegen dispatcher via - * [[CometScalaUDF]]. + * `rlike` uses a plan-time whitelist ([[org.apache.comet.expressions.CometRegex]]) to decide the + * engine. A `UTF8_BINARY` literal pattern that the analyzer proves equivalent to Java regex runs + * natively by default. Every other case stays on the JVM codegen dispatcher (Spark's own + * `doGenCode` inside the Comet pipeline) unless the user sets + * `spark.comet.expression.RLike.allowIncompatible=true`, which forces the native Rust path for + * any non-null string literal, including patterns the analyzer cannot prove equivalent. A + * non-literal or NULL pattern always stays on the dispatcher. Falls through to Spark when the + * dispatcher is disabled. */ -object CometRLike extends CometExpressionSerde[RLike] with NativeOptInAvailable { +object CometRLike + extends CometExpressionSerde[RLike] + with NativeOptInAvailable + with CometTypeShim { + + override def getCompatibleNotes(): Seq[String] = + Seq( + "When the pattern is a `UTF8_BINARY` literal that uses only constructs the plan-time " + + "analyzer proves equivalent to Java regex (ASCII literals, simple character classes, " + + "ordinary greedy quantifiers, capturing / non-capturing groups, and alternation), " + + "Comet evaluates `rlike` natively by default.") override def getIncompatibleReasons(): Seq[String] = Seq("Uses Rust regexp engine, which has different behavior to Java regexp engine") - private def nativeApplicable(expr: RLike): Boolean = expr.right match { - case Literal(_, DataTypes.StringType) => true - case _ => false + private def literalPattern(expr: RLike): Option[String] = expr.right match { + case Literal(v: UTF8String, _: StringType) => Some(v.toString) + case _ => None } - override def getSupportLevel(expr: RLike): SupportLevel = - if (!CometConf.isExprAllowIncompat(getExprConfigName(expr)) && nativeApplicable(expr)) { + private def hasNonDefaultCollation(expr: RLike): Boolean = + hasNonDefaultStringCollation(expr.left.dataType) || + hasNonDefaultStringCollation(expr.right.dataType) + + private def nativeApplicable(expr: RLike): Boolean = literalPattern(expr).isDefined + + private def provablyCompatible(expr: RLike): Boolean = + !hasNonDefaultCollation(expr) && + literalPattern(expr).exists { p => + CometRegex.supportLevel(p, RegexFlavor.RLike).isInstanceOf[Compatible] + } + + override def getSupportLevel(expr: RLike): SupportLevel = { + val allowIncompat = CometConf.isExprAllowIncompat(getExprConfigName(expr)) + if (provablyCompatible(expr) || allowIncompat) { + Compatible() + } else if (nativeApplicable(expr)) { Compatible(nativeOptIn = Some(NativeOptIn(CometConf.getExprAllowIncompatConfigKey(getExprConfigName(expr))))) } else { Compatible() } + } override def convert(expr: RLike, inputs: Seq[Attribute], binding: Boolean): Option[Expr] = { - if (CometConf.isExprAllowIncompat(getExprConfigName(expr)) && nativeApplicable(expr)) { - // Native path: the Rust regexp engine has different semantics from Java regexp. + val allowIncompat = CometConf.isExprAllowIncompat(getExprConfigName(expr)) + if (provablyCompatible(expr) || (allowIncompat && nativeApplicable(expr))) { return createBinaryExpr( expr, expr.left, @@ -394,8 +425,8 @@ object CometRLike extends CometExpressionSerde[RLike] with NativeOptInAvailable binding, (builder, binaryExpr) => builder.setRlike(binaryExpr)) } - // Default: route through the codegen dispatcher so Spark's own doGenCode runs inside the Comet - // pipeline. Falls back to Spark when the dispatcher is disabled. + // Out-of-subset literal, non-literal, NULL pattern, or non-default collation: run Spark's + // own doGenCode inside the Comet pipeline. Falls back to Spark when the dispatcher is off. CometScalaUDF.emitJvmCodegenDispatch(expr, inputs, binding) } } diff --git a/spark/src/test/resources/sql-tests/expressions/string/rlike_auto_native.sql b/spark/src/test/resources/sql-tests/expressions/string/rlike_auto_native.sql new file mode 100644 index 00000000000..e93b18491e9 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/string/rlike_auto_native.sql @@ -0,0 +1,44 @@ +-- 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. + +-- Default settings: in-subset literal patterns run on the native Rust path. +-- Routing is asserted in CometRegExpJvmSuite / CometRegexParitySuite; this file +-- only checks result equality with Spark. + +statement +CREATE TABLE test_rlike_auto(s string) USING parquet + +statement +INSERT INTO test_rlike_auto VALUES ('hello'), ('12345'), (''), (NULL), ('Hello World'), ('abc123'), ('aa'), ('ab'), ('foo'), ('bar'), ('a+b') + +query +SELECT s RLIKE 'abc[0-9]+' FROM test_rlike_auto + +query +SELECT s RLIKE '[a-zA-Z_][a-zA-Z0-9_]*' FROM test_rlike_auto + +query +SELECT s RLIKE '(foo|bar){1,3}' FROM test_rlike_auto + +query +SELECT s RLIKE 'a\+b' FROM test_rlike_auto + +query +SELECT s RLIKE '' FROM test_rlike_auto + +query +SELECT 'hello' RLIKE '[a-z]+', '12345' RLIKE '[0-9]+', '' RLIKE '', NULL RLIKE 'a' diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala index 80f92c4b7f5..5dc977bae59 100644 --- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala @@ -3478,17 +3478,21 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { assert(Compatible().nativeOptIn.isEmpty) } - test("RLike literal pattern shows native opt-in, non-literal does not") { + test("RLike out-of-subset literal shows native opt-in, in-subset and non-literal do not") { withTable("t") { spark.sql("create table t(s string, p string) using parquet") spark.sql("insert into t values ('abc','a.*'), ('xyz','z')") - val lit = spark.sql("select s rlike 'a.*' as r from t") + val unsafeLit = spark.sql("select s rlike 'a.*' as r from t") + val safeLit = spark.sql("select s rlike 'abc[0-9]+' as r from t") val nonLit = spark.sql("select s rlike p as r from t") - val explainLit = - new ExtendedExplainInfo().generateExtendedInfo(lit.queryExecution.executedPlan) + val explainUnsafe = + new ExtendedExplainInfo().generateExtendedInfo(unsafeLit.queryExecution.executedPlan) + val explainSafe = + new ExtendedExplainInfo().generateExtendedInfo(safeLit.queryExecution.executedPlan) val explainNonLit = new ExtendedExplainInfo().generateExtendedInfo(nonLit.queryExecution.executedPlan) - assert(explainLit.contains("native implementation of RLike")) + assert(explainUnsafe.contains("native implementation of RLike")) + assert(!explainSafe.contains("native implementation of RLike")) assert(!explainNonLit.contains("native implementation of RLike")) } } diff --git a/spark/src/test/scala/org/apache/comet/CometRegExpJvmSuite.scala b/spark/src/test/scala/org/apache/comet/CometRegExpJvmSuite.scala index 10f6ad27b49..52ae52b6b03 100644 --- a/spark/src/test/scala/org/apache/comet/CometRegExpJvmSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometRegExpJvmSuite.scala @@ -23,8 +23,10 @@ import org.apache.spark.sql.CometTestBase import org.apache.spark.sql.comet.{CometFilterExec, CometProjectExec} import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper -// No per-expression `allowIncompatible` is set, so the regex family runs through the codegen -// dispatcher (Spark's own code, enabled by default) rather than the native rust path. +import org.apache.comet.CometSparkSessionExtensions.isSpark40Plus + +// Regex expressions other than in-subset `rlike` run through the codegen dispatcher by default +// (Spark's own code, enabled by default) rather than the native rust path. class CometRegExpJvmSuite extends CometTestBase with AdaptiveSparkPlanHelper { // Patterns that the Rust regex crate cannot handle. Using one of these proves @@ -128,12 +130,6 @@ class CometRegExpJvmSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } - test("rlike: null literal pattern falls back to Spark") { - withSubjects("a", "b", null) { - checkSparkAnswer(sql("SELECT s rlike CAST(NULL AS STRING) FROM t")) - } - } - test("rlike: invalid pattern falls back to Spark") { withSubjects("a") { val ex = intercept[Throwable](sql("SELECT s rlike '[' FROM t").collect()) @@ -171,6 +167,351 @@ class CometRegExpJvmSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } + private def withRLikeExplain(f: => Unit): Unit = { + withSQLConf( + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "true", + CometConf.COMET_EXPLAIN_CODEGEN_ENABLED.key -> "true", + CometConf.COMET_EXTENDED_EXPLAIN_FORMAT.key -> + CometConf.COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE)(f) + } + + private def explainOf(df: org.apache.spark.sql.DataFrame): String = + new ExtendedExplainInfo().generateExtendedInfo(df.queryExecution.executedPlan) + + private def assertSparkRegexError(query: String): Unit = { + def collectError(): Throwable = + intercept[Throwable](sql(query).collect()) + + def chain(ex: Throwable): List[Throwable] = + Iterator.iterate(ex)(_.getCause).takeWhile(_ != null).toList + + var sparkEx: Throwable = null + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + sparkEx = collectError() + } + val cometEx = collectError() + val sparkMsgs = chain(sparkEx).flatMap(e => Option(e.getMessage)).mkString("\n") + val cometMsgs = chain(cometEx).flatMap(e => Option(e.getMessage)).mkString("\n") + assert( + sparkMsgs.toLowerCase.contains("unclosed") || sparkMsgs.contains("PatternSyntax") || + sparkMsgs.toLowerCase.contains("regex"), + s"Spark error did not look like a regex syntax error: $sparkMsgs") + assert( + cometMsgs.toLowerCase.contains("unclosed") || cometMsgs.contains("PatternSyntax") || + cometMsgs.toLowerCase.contains("regex"), + s"Comet error did not look like a regex syntax error: $cometMsgs") + val sparkTypes = chain(sparkEx).map(_.getClass.getName) + val cometTypes = chain(cometEx).map(_.getClass.getName) + assert( + sparkTypes.exists(cometTypes.contains), + s"Comet exception types $cometTypes did not share a type with Spark $sparkTypes") + } + + test("rlike: safe literal pattern takes the native path by default") { + withRLikeExplain { + withSubjects("abc123", "xyz", null, "abc") { + val df = sql("SELECT s, s rlike 'abc[0-9]+' FROM t") + checkSparkAnswerAndOperator(df) + val explain = explainOf(df) + assert( + !explain.contains("JVM codegen dispatcher: rlike"), + s"expected native path for in-subset pattern, got:\n$explain") + } + } + } + + test("rlike: Rust class set operations stay on the dispatcher") { + withRLikeExplain { + withSubjects("~", "a", "b", "x", null) { + Seq("[a~~b]", "[^a~~b]").foreach { pat => + val df = sql(s"SELECT s, s rlike '$pat' FROM t") + checkSparkAnswerAndOperator(df) + assert( + explainOf(df).contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher for $pat, got:\n${explainOf(df)}") + } + } + withSubjects("b", "a", "z", "-", null) { + val df = sql("SELECT s, s rlike '[a-z--b]' FROM t") + checkSparkAnswerAndOperator(df) + assert( + explainOf(df).contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher for [a-z--b], got:\n${explainOf(df)}") + } + } + } + + test("rlike: leading-bracket class ranges stay on the dispatcher") { + withRLikeExplain { + withSubjects("_", "-", "]", "a", "z", null) { + Seq("[]-a]", "[^]-a]").foreach { pat => + val df = sql(s"SELECT s, s rlike '$pat' FROM t") + checkSparkAnswerAndOperator(df) + assert( + explainOf(df).contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher for [$pat], got:\n${explainOf(df)}") + } + } + } + } + + test("rlike: raw [ range endpoint stays on the dispatcher and preserves Spark error") { + withRLikeExplain { + withSubjects("@", "[", "A", null) { + Seq("[@-[]", "[^@-[]").foreach { pat => + val query = s"SELECT s, s rlike '$pat' FROM t" + val df = sql(query) + assert( + explainOf(df).contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher for $pat, got:\n${explainOf(df)}") + assertSparkRegexError(query) + } + } + } + } + + test("rlike: over-budget counted repetition stays on the dispatcher") { + withRLikeExplain { + withSubjects("a", "aaa", null) { + val df = sql("SELECT s, s rlike 'a{1000000}' FROM t") + checkSparkAnswerAndOperator(df) + assert( + explainOf(df).contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher for a{1000000}, got:\n${explainOf(df)}") + } + withSubjects(";", "x", "xx", null) { + val df = sql("SELECT s, s rlike '[^;]{20000}' FROM t") + checkSparkAnswerAndOperator(df) + assert( + explainOf(df).contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher for [^;]{20000}, got:\n${explainOf(df)}") + } + withSubjects("a", "aaa", null) { + val df = sql("SELECT s, s rlike '(a{100}){100}' FROM t") + checkSparkAnswerAndOperator(df) + assert( + explainOf(df).contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher for (a{100}){100}, got:\n${explainOf(df)}") + } + withSubjects("", "x", ";" * 256, null) { + val pat = "(([^;]{256}){0,}){256}" + val df = sql(s"SELECT s, s rlike '$pat' FROM t") + checkSparkAnswerAndOperator(df) + assert( + explainOf(df).contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher for $pat, got:\n${explainOf(df)}") + } + withSubjects("a", "b", null) { + val nested = "(" * 33 + "a" + ")" * 33 + val df = sql(s"SELECT s, s rlike '$nested' FROM t") + checkSparkAnswerAndOperator(df) + assert( + explainOf(df).contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher for 33 nested groups, got:\n${explainOf(df)}") + } + } + } + + test("rlike: compile-budget boundary stays native") { + withRLikeExplain { + withSubjects("a", "aaa", null) { + val pat = "(a{64}){64}" + val df = sql(s"SELECT s, s rlike '$pat' FROM t") + checkSparkAnswerAndOperator(df) + assert( + !explainOf(df).contains("JVM codegen dispatcher: rlike"), + s"expected native path for expansion-4096 pattern, got:\n${explainOf(df)}") + } + } + } + + test("rlike: exact-zero counted repetitions stay native") { + withRLikeExplain { + withSubjects("", "x", ";" * 256, null) { + Seq("(([^;]{256}){0}){256}", "(([^;]{256}){0,0}){256}").foreach { pat => + val df = sql(s"SELECT s, s rlike '$pat' FROM t") + checkSparkAnswerAndOperator(df) + assert( + !explainOf(df).contains("JVM codegen dispatcher: rlike"), + s"expected native path for exact-zero pattern $pat, got:\n${explainOf(df)}") + } + } + } + } + + test("rlike: aggregate expansion budget stays on the dispatcher") { + withRLikeExplain { + withSubjects("a", "aaa", null) { + Seq("a{256}" * 17, "a{0}" * 4097).foreach { pat => + val df = sql(s"SELECT s, s rlike '$pat' FROM t") + checkSparkAnswerAndOperator(df) + assert( + explainOf(df).contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher for aggregate over-budget pattern, got:\n${explainOf(df)}") + } + } + } + } + + test("rlike: unsafe literal pattern stays on the JVM dispatcher by default") { + withRLikeExplain { + withSubjects("abc123", "no digits", null) { + val df = sql("SELECT s, s rlike '\\\\d+' FROM t") + checkSparkAnswerAndOperator(df) + val explain = explainOf(df) + assert( + explain.contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher path for out-of-subset pattern, got:\n$explain") + } + } + } + + test("rlike: unsafe but Rust-accepted pattern is native after opt-in") { + withRLikeExplain { + withSQLConf(CometConf.getExprAllowIncompatConfigKey("RLike") -> "true") { + withSubjects("abc123", "no digits", null) { + val df = sql("SELECT s, s rlike '\\\\d+' FROM t") + checkSparkAnswerAndOperator(df) + val explain = explainOf(df) + assert( + !explain.contains("JVM codegen dispatcher: rlike"), + s"expected native path after allowIncompatible, got:\n$explain") + } + } + } + } + + test("rlike: non-literal pattern stays on the dispatcher") { + withRLikeExplain { + withTable("t") { + sql("CREATE TABLE t (s STRING, p STRING) USING parquet") + sql("INSERT INTO t VALUES ('abc123', 'abc[0-9]+'), ('xyz', 'xyz')") + val df = sql("SELECT s, s rlike p FROM t") + checkSparkAnswerAndOperator(df) + val explain = explainOf(df) + assert( + explain.contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher path for non-literal pattern, got:\n$explain") + } + } + } + + test("rlike: null literal pattern stays on the dispatcher") { + // NullPropagation would replace `s rlike NULL` with a null literal before serde sees it. + withRLikeExplain { + withSQLConf( + "spark.sql.optimizer.excludedRules" -> + "org.apache.spark.sql.catalyst.optimizer.NullPropagation") { + withSubjects("a", "b", null) { + val df = sql("SELECT s rlike CAST(NULL AS STRING) FROM t") + checkSparkAnswerAndOperator(df) + val explain = explainOf(df) + assert( + explain.contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher path for null literal pattern, got:\n$explain") + } + } + } + } + + test("rlike: unsafe pattern falls back to Spark when the dispatcher is disabled") { + withSubjects("abc123", "no digits", null) { + withSQLConf(CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "false") { + checkSparkAnswerAndFallbackReason( + sql("SELECT s, s rlike '\\\\d+' FROM t"), + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key + "=false") + } + } + } + + test("rlike: in-subset pattern stays native when the dispatcher is disabled") { + withSQLConf( + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "false", + CometConf.COMET_EXPLAIN_CODEGEN_ENABLED.key -> "true", + CometConf.COMET_EXTENDED_EXPLAIN_FORMAT.key -> + CometConf.COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE) { + withSubjects("abc123", "xyz", null, "abc") { + val df = sql("SELECT s, s rlike 'abc[0-9]+' FROM t") + checkSparkAnswerAndOperator(df) + val explain = explainOf(df) + assert( + !explain.contains("JVM codegen dispatcher: rlike"), + s"expected native path with dispatcher disabled, got:\n$explain") + assert( + !explain.contains(CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key), + s"in-subset rlike must not fall back when the dispatcher is off, got:\n$explain") + } + } + } + + test("rlike: invalid pattern preserves Spark exception type and message") { + withSubjects("a") { + assertSparkRegexError("SELECT s rlike '[' FROM t") + } + } + + test("rlike: Java-only pattern with allowIncompatible keeps existing opt-in behavior") { + // Pre-existing: opt-in sends any literal to native. Rust cannot compile lookaround, so + // native plan construction fails. This PR does not add a fallback for that case. + withSQLConf(CometConf.getExprAllowIncompatConfigKey("RLike") -> "true") { + withSubjects("foobar") { + val ex = intercept[Throwable](sql(s"SELECT s rlike '$lookahead' FROM t").collect()) + val msgs = + Iterator + .iterate(ex)(_.getCause) + .takeWhile(_ != null) + .flatMap(e => Option(e.getMessage)) + .mkString("\n") + assert( + msgs.toLowerCase.contains("pattern") || msgs.toLowerCase.contains("regex") || + msgs.toLowerCase.contains("look"), + s"expected native compile failure for lookaround under opt-in, got: $msgs") + } + } + } + + test("rlike: UTF8_BINARY safe literal is native on Spark 4") { + assume(isSpark40Plus) + withRLikeExplain { + withSubjects("abc123", "xyz") { + val df = sql("SELECT s rlike 'abc[0-9]+' FROM t") + checkSparkAnswerAndOperator(df) + val explain = explainOf(df) + assert( + !explain.contains("JVM codegen dispatcher: rlike"), + s"expected native path for UTF8_BINARY, got:\n$explain") + } + } + } + + test("rlike: non-default collation on the subject stays on the dispatcher") { + assume(isSpark40Plus) + withRLikeExplain { + withSubjects("abc123", "ABC123", null) { + val df = sql("SELECT CAST(s AS STRING COLLATE UTF8_LCASE) rlike 'abc[0-9]+' FROM t") + checkSparkAnswerAndOperator(df) + val explain = explainOf(df) + assert( + explain.contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher for collated subject, got:\n$explain") + } + } + } + + test("rlike: non-default collation on the pattern stays on the dispatcher") { + assume(isSpark40Plus) + withRLikeExplain { + withSubjects("abc123", "xyz", null) { + val df = sql("SELECT s rlike CAST('abc[0-9]+' AS STRING COLLATE UTF8_LCASE) FROM t") + checkSparkAnswerAndOperator(df) + val explain = explainOf(df) + assert( + explain.contains("JVM codegen dispatcher: rlike"), + s"expected dispatcher for collated pattern, got:\n$explain") + } + } + } + // ========== regexp_extract tests ========== test("regexp_extract: basic group extraction") { diff --git a/spark/src/test/scala/org/apache/comet/CometRegexParitySuite.scala b/spark/src/test/scala/org/apache/comet/CometRegexParitySuite.scala new file mode 100644 index 00000000000..5d00050d5db --- /dev/null +++ b/spark/src/test/scala/org/apache/comet/CometRegexParitySuite.scala @@ -0,0 +1,174 @@ +/* + * 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. + */ + +package org.apache.comet + +import org.apache.spark.sql.CometTestBase +import org.apache.spark.sql.Row +import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper +import org.apache.spark.sql.types.{StringType, StructField, StructType} + +import org.apache.comet.expressions.{CometRegex, RegexFlavor} +import org.apache.comet.serde.Compatible + +/** + * Differential corpus: every pattern the plan-time analyzer admits is evaluated on Spark (Java + * `Pattern`) and on Comet's native Rust path, and the results must match. Routing is asserted via + * extended explain so a dispatcher fallback cannot silently satisfy the comparison. + */ +class CometRegexParitySuite extends CometTestBase with AdaptiveSparkPlanHelper { + + // One pattern per whitelist production, plus the lexer-boundary cases. + private val admittedPatterns: Seq[String] = Seq( + "", + "abc", + "abc[0-9]+", + "[a-zA-Z_][a-zA-Z0-9_]*", + "(foo|bar){1,3}", + "a\\+b", + "[0-9_]", + "[^0-9]", + "(?:foo|bar)", + "(foo)", + "abc|def", + "a*", + "a+", + "a?", + "a{2}", + "a{2,}", + "a{2,4}", + "[(?=]", + "\\(\\?=", + "\\\\d", + "[.]", + "a|", + "|a", + "[a-]", + "[-a]", + "[a\\-z]", + "a b", + "(?:(?:foo)|bar)", + "(ab)+", + "[@-\\[]") + + private val subjects: Seq[String] = Seq( + "abc", + "abc123", + "ABC", + "", + null, + "foo", + "bar", + "foobar", + "a+b", + "\\d", + "αβγ", + "١٢٣", + "😀", + "e\u0301", + "\n", + "\r", + "\r\n", + "\u0085", + "\u2028", + "\u2029", + "\nabc", + "abc\n", + "\nabc\n", + "(?=", + ".", + "aa", + "aaaa", + "-", + "z", + "a b", + "ab", + "abab", + "b", + "@", + "[", + "A") + + private def sqlLiteral(pattern: String): String = + pattern.replace("\\", "\\\\").replace("'", "''") + + private def withSubjectTable(values: Seq[String])(f: => Unit): Unit = { + val schema = StructType(Seq(StructField("s", StringType, nullable = true))) + val rows = spark.sparkContext.parallelize(values.map(Row(_))) + val df = spark.createDataFrame(rows, schema) + df.createOrReplaceTempView("t") + f + } + + private def explainInfo(df: org.apache.spark.sql.DataFrame): String = + new ExtendedExplainInfo().generateExtendedInfo(df.queryExecution.executedPlan) + + test("analyzer admits every corpus pattern") { + admittedPatterns.foreach { p => + val level = CometRegex.supportLevel(p, RegexFlavor.RLike) + assert(level.isInstanceOf[Compatible], s"corpus pattern must be Compatible: [$p] -> $level") + } + } + + test("native rlike matches Spark for every admitted pattern") { + withSQLConf( + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "true", + CometConf.COMET_EXPLAIN_CODEGEN_ENABLED.key -> "true", + CometConf.COMET_EXTENDED_EXPLAIN_FORMAT.key -> + CometConf.COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE) { + withSubjectTable(subjects) { + admittedPatterns.foreach { pattern => + val lit = sqlLiteral(pattern) + val projected = sql(s"SELECT s, s rlike '$lit' FROM t") + checkSparkAnswerAndOperator(projected) + val projectedExplain = explainInfo(projected) + assert( + !projectedExplain.contains("JVM codegen dispatcher: rlike"), + s"expected native path for [$pattern], got:\n$projectedExplain") + + val filtered = sql(s"SELECT s FROM t WHERE s rlike '$lit'") + checkSparkAnswerAndOperator(filtered) + val filteredExplain = explainInfo(filtered) + assert( + !filteredExplain.contains("JVM codegen dispatcher: rlike"), + s"expected native path for filter [$pattern], got:\n$filteredExplain") + } + } + } + } + + test("native rlike matches Spark across multiple Arrow batches") { + withSQLConf( + CometConf.COMET_BATCH_SIZE.key -> "64", + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "true", + CometConf.COMET_EXPLAIN_CODEGEN_ENABLED.key -> "true", + CometConf.COMET_EXTENDED_EXPLAIN_FORMAT.key -> + CometConf.COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE) { + val many = (0 until 5000).map(i => if (i % 7 == 0) null else s"row_${i}_abc123") + withSubjectTable(many) { + val df = sql("SELECT s, s rlike 'abc[0-9]+' FROM t") + checkSparkAnswerAndOperator(df) + val explain = explainInfo(df) + assert( + !explain.contains("JVM codegen dispatcher: rlike"), + s"expected native path for multi-batch rlike, got:\n$explain") + } + } + } +} diff --git a/spark/src/test/scala/org/apache/comet/expressions/CometRegexSuite.scala b/spark/src/test/scala/org/apache/comet/expressions/CometRegexSuite.scala new file mode 100644 index 00000000000..70a7115a212 --- /dev/null +++ b/spark/src/test/scala/org/apache/comet/expressions/CometRegexSuite.scala @@ -0,0 +1,177 @@ +/* + * 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. + */ + +package org.apache.comet.expressions + +import org.scalatest.funsuite.AnyFunSuite + +import org.apache.comet.serde.{Compatible, Incompatible} + +class CometRegexSuite extends AnyFunSuite { + + private def assertCompatible(pattern: String): Unit = { + val level = CometRegex.supportLevel(pattern, RegexFlavor.RLike) + assert(level.isInstanceOf[Compatible], s"expected Compatible for [$pattern], got $level") + } + + private def assertIncompatible(pattern: String): Unit = { + val level = CometRegex.supportLevel(pattern, RegexFlavor.RLike) + assert(level.isInstanceOf[Incompatible], s"expected Incompatible for [$pattern], got $level") + } + + test("admits ASCII literals, classes, quantifiers, groups, and alternation") { + Seq( + "", + "abc", + "abc[0-9]+", + "[a-zA-Z_][a-zA-Z0-9_]*", + "(foo|bar){1,3}", + "a\\+b", + "[0-9_]", + "[^0-9]", + "(?:foo)", + "(foo)", + "abc|def", + "a*", + "a+", + "a?", + "a{2}", + "a{2,}", + "a{2,4}", + "a|", + "|a", + "[a-]", + "[-a]", + "[a\\-z]", + "a b", + "(?:(?:foo)|bar)", + "(ab)+", + "[a~b]", + "(a{2}){3}").foreach(assertCompatible) + } + + test("admits lexer-boundary patterns that a substring search would misclassify") { + Seq( + "[(?=]", // class of literals, not lookahead + "\\(\\?=", // escaped literal `(?=` + "\\\\d", // literal backslash plus `d` + "[.]" // class-local literal dot + ).foreach(assertCompatible) + } + + test("rejects constructs that diverge from Java regex or are unrecognized") { + Seq( + "\\d+", + "\\w+", + ".", + "(?i)abc", + "(?m)^abc$", + "(foo)\\1", + "foo(?=bar)", + "(?<=foo)bar", + "a*+", + "(?>abc)", + "\\p{L}+", + "[a-z&&[^aeiou]]", + "\\u0041", + "\\012", + "^abc", + "abc$", + "abc.", + "[", + "\\s+", + "\\b", + "\\B", + "a+?", + "a*?", + "(?foo)", + "\\n", + "\\t", + "café", + "你好").foreach(assertIncompatible) + } + + test("rejects scanner edge cases the whitelist must not silently admit") { + Seq( + "\\A", // beginning of input; Java-only relative to the Rust crate + "\\Z", // end of input, before final line terminator + "\\z", // absolute end of input + "\\G", // end of previous match + "\\W", + "\\S", + "\\D", + "a{2,1}", // inverted counted range + "{,4}", // missing lower bound; `{` is not a valid atom + "a{2", // unclosed counted quantifier + "(", // unclosed group + "(abc", + "[z-a]", // inverted character-class range + "[[a]]", // nested class + "\\x41", // hex escape + "\\Qabc\\E" // quoted span + ).foreach(assertIncompatible) + } + + test("rejects Rust-only character-class set operations") { + Seq("[a~~b]", "[a-z--b]", "[^a~~b]", "[^a-z--b]", "[a&&b]").foreach(assertIncompatible) + } + + test("rejects unescaped ] used as a character-class atom or range endpoint") { + Seq("[]-a]", "[^]-a]", "[]]", "[^]]").foreach(assertIncompatible) + assertCompatible("[\\]]") + assertCompatible("[a\\]]") + } + + test("rejects unescaped [ used as a character-class range endpoint") { + Seq("[@-[]", "[^@-[]").foreach(assertIncompatible) + assertCompatible("[@-\\[]") + } + + test("rejects counted or nested patterns that can exceed the Rust compile budget") { + Seq( + "a{1000000}", + "[^;]{20000}", + "a{257}", + "(a{100}){100}", + "(([^;]{256}){0,}){256}", + "(" * 33 + "a" + ")" * 33) + .foreach(assertIncompatible) + assertCompatible("a{256}") + assertCompatible("(a{2}){3}") + assertCompatible("(([^;]{256}){0}){256}") + assertCompatible("(([^;]{256}){0,0}){256}") + assertCompatible("(" * 32 + "a" + ")" * 32) + } + + test("rejects aggregate patterns that exceed the Rust compile budget") { + val maxExpansion = 4096 + assertCompatible("a{256}" * 16) + assertCompatible("(a{64}){64}") + assertCompatible("a{0}" * maxExpansion) + Seq( + "a{256}" * 17, + "(a)" * (maxExpansion + 1), + List.fill(maxExpansion + 1)("a").mkString("|"), + "a{0}" * (maxExpansion + 1), + "a{0,}" * (maxExpansion + 1), + "a{0,0}" * (maxExpansion + 1), + "a{256}" * 16 + "a{0}") + .foreach(assertIncompatible) + } +} diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometRegExpBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometRegExpBenchmark.scala index 5a6741dc2a8..21a4b42024d 100644 --- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometRegExpBenchmark.scala +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometRegExpBenchmark.scala @@ -34,11 +34,12 @@ import org.apache.comet.CometConf case class RegExpPattern(name: String, pattern: String) /** - * Benchmark `rlike` across all execution modes: - * - Spark - * - Comet (Scan only) - * - Comet (Scan + Exec, native Rust regex) - * - Comet (Scan + Exec, JVM-side java.util.regex) + * Benchmark `rlike` across execution modes. + * + * In-subset patterns (proved Java-equivalent by [[org.apache.comet.expressions.CometRegex]]) take + * the native path by default, so they are measured as Spark / Scan / Native. Out-of-subset + * patterns that Rust can still compile (`\d+`) keep a four-way comparison: default is the JVM + * dispatcher, and `allowIncompatible` selects native. * * To run: * {{{ @@ -50,28 +51,44 @@ case class RegExpPattern(name: String, pattern: String) */ object CometRegExpBenchmark extends CometBenchmarkBase { - // Patterns chosen to span common rlike shapes. Avoid Java-only constructs - // that the native (Rust) path cannot accept, since those would be skipped - // rather than benchmarked in the native case. - private val patterns = List( + // Analyzer-admitted patterns. Default Comet exec is already native, so do not add a + // "JVM regex" case: it would silently measure the native path. + private val inSubsetPatterns = List( RegExpPattern("character_class", "[0-9]+"), - RegExpPattern("anchored", "^[0-9]"), RegExpPattern("alternation", "abc|def|ghi"), RegExpPattern("multi_class", "[a-zA-Z][0-9]+"), RegExpPattern("repetition", "(ab){2,}")) + // Analyzer-rejected, Rust-accepted. Default exec is the JVM dispatcher; opt-in is native. + // Input data is ASCII (REPEAT of numeric strings) so `\d` vs `[0-9]` does not change hits. + private val outOfSubsetPatterns = List(RegExpPattern("digit_class_shorthand", "\\d+")) + + // Spark's SQL parser consumes one backslash layer and treats `'` as the + // string delimiter. Escape so the regex engine sees the intended pattern. + private def sqlRegexLiteral(pattern: String): String = + pattern.replace("\\", "\\\\").replace("'", "''") + + private def rlikeQuery(pattern: String): String = + s"select c1 rlike '${sqlRegexLiteral(pattern)}' from parquetV1Table" + override def runCometBenchmark(mainArgs: Array[String]): Unit = { - runBenchmarkWithTable("rlike modes", 1024) { v => + runBenchmarkWithTable("rlike modes", 1024 * 1024) { v => withTempPath { dir => withTempTable("parquetV1Table") { prepareTable( dir, spark.sql(s"SELECT REPEAT(CAST(value AS STRING), 10) AS c1 FROM $tbl")) - patterns.foreach { p => - val query = s"select c1 rlike '${p.pattern}' from parquetV1Table" + inSubsetPatterns.foreach { p => + val query = rlikeQuery(p.pattern) runBenchmark(p.name) { - runRLikeModes(p.name, v, query) + runInSubsetModes(p.name, v, query) + } + } + outOfSubsetPatterns.foreach { p => + val query = rlikeQuery(p.pattern) + runBenchmark(p.name) { + runOutOfSubsetModes(p.name, v, query) } } } @@ -79,16 +96,18 @@ object CometRegExpBenchmark extends CometBenchmarkBase { } } - /** Runs all four modes for a single rlike query. */ - private def runRLikeModes(name: String, cardinality: Long, query: String): Unit = { - val benchmark = new Benchmark(name, cardinality, output = output) + private val baseExec: Map[String, String] = Map( + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + "spark.sql.optimizer.excludedRules" -> + "org.apache.spark.sql.catalyst.optimizer.ConstantFolding") + private def addSparkAndScan(benchmark: Benchmark, query: String): Unit = { benchmark.addCase("Spark") { _ => withSQLConf(CometConf.COMET_ENABLED.key -> "false") { spark.sql(query).noop() } } - benchmark.addCase("Comet (Scan)") { _ => withSQLConf( CometConf.COMET_ENABLED.key -> "true", @@ -96,27 +115,35 @@ object CometRegExpBenchmark extends CometBenchmarkBase { spark.sql(query).noop() } } + } - val baseExec = Map( - CometConf.COMET_ENABLED.key -> "true", - CometConf.COMET_EXEC_ENABLED.key -> "true", - "spark.sql.optimizer.excludedRules" -> - "org.apache.spark.sql.catalyst.optimizer.ConstantFolding") + /** Spark / Scan / Native (the new default for in-subset patterns). */ + private def runInSubsetModes(name: String, cardinality: Long, query: String): Unit = { + val benchmark = new Benchmark(name, cardinality, output = output) + addSparkAndScan(benchmark, query) + benchmark.addCase("Comet (Exec, native Rust regex)") { _ => + withSQLConf(baseExec.toSeq: _*) { + spark.sql(query).noop() + } + } + benchmark.run() + } + /** Spark / Scan / Native-on-opt-in / JVM-dispatcher-by-default. */ + private def runOutOfSubsetModes(name: String, cardinality: Long, query: String): Unit = { + val benchmark = new Benchmark(name, cardinality, output = output) + addSparkAndScan(benchmark, query) benchmark.addCase("Comet (Exec, native Rust regex)") { _ => val configs = baseExec ++ Map(CometConf.getExprAllowIncompatConfigKey("RLike") -> "true") withSQLConf(configs.toSeq: _*) { spark.sql(query).noop() } } - benchmark.addCase("Comet (Exec, JVM regex)") { _ => - // The codegen dispatcher is enabled by default, so no extra config is needed. withSQLConf(baseExec.toSeq: _*) { spark.sql(query).noop() } } - benchmark.run() } }