viirya commented on code in PR #59879:
URL: https://github.com/apache/spark/pull/59879#discussion_r4428905945
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -670,6 +670,20 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val REGEX_PATTERN_COST_ESTIMATE_VERSION =
+ buildConf("spark.sql.optimizer.regexPatternCostEstimateVersion")
+ .internal()
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .doc(s"How ${AVOID_DOUBLE_FILTER_EVAL.key} decides whether a LIKE or RLIKE
expression is " +
+ "expensive. 0: every LIKE and RLIKE expression is expensive, as in 4.2.0
for a subject " +
+ "that is not a literal. 1: only one whose pattern is likely to be
expensive to match. A " +
+ "streaming query keeps the version recorded in its checkpoint, and one
restarted from a " +
+ "checkpoint without it uses 0.")
+ .version("4.2.2")
+ .intConf
+ .checkValues(Set(0, 1))
Review Comment:
Agreed, a downgrade should not fail the query. It is now `checkValue(_ >= 0,
...)`, and any version other than `0` runs the current estimate.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala:
##########
@@ -328,16 +332,66 @@ class FilterPushdownSuite extends PlanTest {
// rlike magic is expensive so not pushed.
val correctAnswer = testStringRelation
.where($"a" + $"b" > 10 && $"a" - $"b" < 5)
- .select($"a" + $"b" as "sum", $"a" - $"b" as "diff", $"e".rlike("magic")
as "f")
+ .select($"a" + $"b" as "sum", $"a" - $"b" as "diff",
expensiveRLike("magic") as "f")
.where($"f")
.analyze
comparePlans(optimized, correctAnswer)
}
+ test("SPARK-60632: Push a filter on a LIKE/RLIKE alias with a cheap constant
pattern") {
+ Seq(
+ $"e".rlike("^1"),
+ $"e".rlike("magic"),
+ $"e".rlike("^a.*b"),
+ $"e".like("1_"),
+ $"e".like("%magic%"),
+ // ILIKE, as the optimizer sees it after `ReplaceExpressions`.
+ Like(Lower($"e"), Lower(Literal("%Magic%")), '\\')).foreach { pred =>
+ val originalQuery = testStringRelation
+ .select($"a", pred as "f")
+ .where($"f")
+ .analyze
+
+ val correctAnswer = testStringRelation
+ .where(pred)
+ .select($"a", pred as "f")
+ .analyze
+
+ comparePlans(Optimize.execute(originalQuery), correctAnswer)
+ }
+ }
+
+ test("SPARK-60632: Do not push a filter on a LIKE/RLIKE alias with a
backtracking pattern") {
+ Seq($"e".rlike("(a+)+b"), $"e".rlike(".*a.*b"),
$"e".like("%a%b%c")).foreach { pred =>
+ val originalQuery = testStringRelation
+ .select($"a", pred as "f")
+ .where($"f")
+ .analyze
+
+ comparePlans(Optimize.execute(originalQuery), originalQuery)
+ }
+ }
+
+ test("SPARK-60632: Do not push a filter on a LIKE/RLIKE alias with version 0
of the estimate") {
Review Comment:
Right. I added your two-alias case to the version-0 test, and checked that
it fails when the two `isExpensive` calls in
`splitProjectForExpensiveConditions` go back to `.expensive`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2897,11 +2897,11 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
expensiveConds: Seq[(Expression, AttributeMap[Alias])])
: (LogicalPlan, AttributeSet, Seq[Expression]) = {
val expensiveAliases = AttributeSet(
- aliasMap.collect { case (attr, alias) if alias.child.expensive => attr })
+ aliasMap.collect { case (attr, alias) if isExpensive(alias.child) =>
attr })
// Expensive sub-expressions behind each expensive alias, to spot when two
aliases are the
// same piece of work. Cheap aliases have none, so we skip the (possibly
wide) cheap rest.
val expensivePartsOf = AttributeMap(expensiveAliases.toSeq.map { attr =>
- attr -> aliasMap(attr).child.collect { case e if e.expensive =>
e.canonicalized }.toSet
+ attr -> aliasMap(attr).child.collect { case e if isExpensive(e) =>
e.canonicalized }.toSet
Review Comment:
Done. `splitProjectForExpensiveConditions` reads the conf once and collects
only the regex expressions themselves at version 0, so it no longer runs
`exists` on every node.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/regexpExpressions.scala:
##########
@@ -84,33 +84,212 @@ abstract class StringRegexExpression extends
BinaryExpression
}
}
- override def expensive: Boolean = hasExpensiveChild || _expensiveRegex
-
- // Heuristic, not designed to be perfect. Look for things likely to have
- // back tracking.
- private val detectExpensiveRegexPattern = Pattern.compile("\\+\\*\\{")
-
- private lazy val _expensiveRegex = {
- // A quick heuristic for expensive a pattern is.
- left match {
- case StringLiteral(str) =>
- // If we have a clear start limited back tracking required.
- if (str.startsWith("^") || str.startsWith("\\b")) {
- false
- } else if (detectExpensiveRegexPattern.matcher(str).matches()) {
- // Greedy matching can be tricky.
- true
- } else {
- // Default to pushdown for now.
- false
- }
- case _ =>
- true // per row regex compilation.
+ override def expensive: Boolean = hasExpensiveChild || _expensivePattern
+
+ // Heuristic, not designed to be perfect. A pattern that is not foldable is
compiled for every
+ // row; a foldable one is left to `isExpensivePattern`. Like
`ConstantFolding`, evaluate a fresh
+ // copy so that a stateful pattern keeps its state.
+ private lazy val _expensivePattern = !right.foldable || {
+ try {
+ val str = right.freshCopyIfContainsStatefulExpression().eval()
+ str != null && isExpensivePattern(str.toString)
+ } catch {
+ // E.g. a pattern that `ConstantFolding` left in a conditional branch
because it throws.
+ case NonFatal(_) => true
}
}
+
+ /** Whether matching against the constant pattern `str` is likely to be
expensive. */
+ protected def isExpensivePattern(str: String): Boolean = true
}
private[catalyst] object StringRegexExpression {
+ /**
+ * Whether matching `regex` is likely to be expensive. A failed match tries
every combination of
+ * the lengths that its repeating quantifiers take, so each one costs a
factor of the input
+ * length. `find` also tries every start position, which costs one more
factor unless the regex
+ * is anchored at the start. So two quantifiers are expensive, and three
when the regex is
+ * anchored or must match the whole input, as the one a LIKE pattern
becomes. Repeating a group
+ * that holds a quantifier or an alternation multiplies the factors, so it
is always expensive.
+ *
+ * @param matchesWhole whether `regex` must match the whole input rather
than be found in it.
+ */
+ def isExpensiveRegex(regex: String, matchesWhole: Boolean): Boolean = {
+ val (quantifiers, topLevelAlternation) = scanQuantifiers(regex)
+ val anchored = matchesWhole || (!topLevelAlternation &&
startsWithAnchor(regex))
+ quantifiers >= (if (anchored) 3 else 2)
+ }
+
+ // Leading inline flags, e.g. `(?i)` or `(?s-m)`.
+ private val leadingFlags = """(?:\(\?[a-zA-Z]*(?:-[a-zA-Z]*)?\))*""".r
+
+ // Whether `regex` starts with `^` or `\A`. A leading flag that mentions
MULTILINE, under which
+ // `^` also matches after every line terminator, makes it not anchored.
+ private def startsWithAnchor(regex: String): Boolean = {
+ val flags = leadingFlags.findPrefixOf(regex).getOrElse("")
+ val rest = regex.substring(flags.length)
+ !flags.contains('m') && (rest.startsWith("^") || rest.startsWith("\\A"))
+ }
+
+ // An inline flag, e.g. `(?x)` or `(?ix:`, that turns on COMMENTS mode,
which lets whitespace
+ // and comments separate a group from its quantifier.
+ private val commentsFlag = Pattern.compile("""\(\?[a-zA-Z]*x""")
+
+ // The product of the choices of the uncounted ranges that makes them count
as one quantifier.
+ private val smallRangesLimit = 100
+
+ /**
+ * Counts the repeating quantifiers (`*`, `+`, `?` and `{n,m}`) in a regex
that can backtrack,
+ * and tells whether the regex has a `|` outside any group. Escaped
characters, quoted text,
+ * character classes and the braces of `\p{..}`, `\x{..}` and `\N{..}` are
skipped. The
+ * alternatives of a `|` outside any group are tried one after another, so
each is counted on its
+ * own and the largest count is returned. These quantifiers are not counted:
+ * - a possessive one, which never backtracks;
+ * - one that ends the regex or an alternative of it, possibly inside
groups that end there too,
+ * which a failed match never backtracks into;
+ * - `?` and a `{n,m}` that spans fewer than 10, which add a small factor.
The choices of such
+ * quantifiers multiply, so they count as one quantifier each time their
product reaches 100.
+ * The count is `Int.MaxValue` for a repeated group that holds a quantifier
that can backtrack or
+ * a `|`, for a regex in COMMENTS mode, and for a `[` that is never closed,
since the scan cannot
+ * follow them.
+ *
+ * A streaming query pins `SQLConf.REGEX_PATTERN_COST_ESTIMATE_VERSION` in
its checkpoint, since
+ * a pattern that turns cheap can move a filter below a stateful operator.
So a change that makes
+ * any pattern cheaper needs a new version of the estimate, and the old one
has to be kept.
+ */
+ private def scanQuantifiers(regex: String): (Int, Boolean) = {
+ // The largest count of the alternatives before the current one.
+ var maxCount = 0
+ var count = 0
+ var smallRanges = 1
+ var classDepth = 0
+ var topLevelAlternation = false
+ // Whether each open group holds a choice that a repetition of it would
multiply, innermost
+ // first.
+ var groups = List.empty[Boolean]
+ var i = 0
+ def markGroup(): Unit = groups = groups match {
+ case _ :: outer => true :: outer
+ case Nil => Nil
+ }
+ // Whether nothing but the closing `)` of open groups comes between `j`
and the end of the
+ // regex or a `|` outside any group.
+ def endsAlternative(j: Int): Boolean = {
+ var end = j
+ var depth = groups.length
Review Comment:
Done as you suggested.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/regexpExpressions.scala:
##########
@@ -84,33 +84,212 @@ abstract class StringRegexExpression extends
BinaryExpression
}
}
- override def expensive: Boolean = hasExpensiveChild || _expensiveRegex
-
- // Heuristic, not designed to be perfect. Look for things likely to have
- // back tracking.
- private val detectExpensiveRegexPattern = Pattern.compile("\\+\\*\\{")
-
- private lazy val _expensiveRegex = {
- // A quick heuristic for expensive a pattern is.
- left match {
- case StringLiteral(str) =>
- // If we have a clear start limited back tracking required.
- if (str.startsWith("^") || str.startsWith("\\b")) {
- false
- } else if (detectExpensiveRegexPattern.matcher(str).matches()) {
- // Greedy matching can be tricky.
- true
- } else {
- // Default to pushdown for now.
- false
- }
- case _ =>
- true // per row regex compilation.
+ override def expensive: Boolean = hasExpensiveChild || _expensivePattern
+
+ // Heuristic, not designed to be perfect. A pattern that is not foldable is
compiled for every
+ // row; a foldable one is left to `isExpensivePattern`. Like
`ConstantFolding`, evaluate a fresh
+ // copy so that a stateful pattern keeps its state.
+ private lazy val _expensivePattern = !right.foldable || {
+ try {
+ val str = right.freshCopyIfContainsStatefulExpression().eval()
+ str != null && isExpensivePattern(str.toString)
+ } catch {
+ // E.g. a pattern that `ConstantFolding` left in a conditional branch
because it throws.
+ case NonFatal(_) => true
}
}
+
+ /** Whether matching against the constant pattern `str` is likely to be
expensive. */
+ protected def isExpensivePattern(str: String): Boolean = true
}
private[catalyst] object StringRegexExpression {
+ /**
+ * Whether matching `regex` is likely to be expensive. A failed match tries
every combination of
+ * the lengths that its repeating quantifiers take, so each one costs a
factor of the input
+ * length. `find` also tries every start position, which costs one more
factor unless the regex
+ * is anchored at the start. So two quantifiers are expensive, and three
when the regex is
+ * anchored or must match the whole input, as the one a LIKE pattern
becomes. Repeating a group
+ * that holds a quantifier or an alternation multiplies the factors, so it
is always expensive.
+ *
+ * @param matchesWhole whether `regex` must match the whole input rather
than be found in it.
+ */
+ def isExpensiveRegex(regex: String, matchesWhole: Boolean): Boolean = {
+ val (quantifiers, topLevelAlternation) = scanQuantifiers(regex)
+ val anchored = matchesWhole || (!topLevelAlternation &&
startsWithAnchor(regex))
+ quantifiers >= (if (anchored) 3 else 2)
+ }
+
+ // Leading inline flags, e.g. `(?i)` or `(?s-m)`.
+ private val leadingFlags = """(?:\(\?[a-zA-Z]*(?:-[a-zA-Z]*)?\))*""".r
+
+ // Whether `regex` starts with `^` or `\A`. A leading flag that mentions
MULTILINE, under which
+ // `^` also matches after every line terminator, makes it not anchored.
+ private def startsWithAnchor(regex: String): Boolean = {
+ val flags = leadingFlags.findPrefixOf(regex).getOrElse("")
+ val rest = regex.substring(flags.length)
+ !flags.contains('m') && (rest.startsWith("^") || rest.startsWith("\\A"))
+ }
+
+ // An inline flag, e.g. `(?x)` or `(?ix:`, that turns on COMMENTS mode,
which lets whitespace
+ // and comments separate a group from its quantifier.
+ private val commentsFlag = Pattern.compile("""\(\?[a-zA-Z]*x""")
+
+ // The product of the choices of the uncounted ranges that makes them count
as one quantifier.
+ private val smallRangesLimit = 100
+
+ /**
+ * Counts the repeating quantifiers (`*`, `+`, `?` and `{n,m}`) in a regex
that can backtrack,
+ * and tells whether the regex has a `|` outside any group. Escaped
characters, quoted text,
+ * character classes and the braces of `\p{..}`, `\x{..}` and `\N{..}` are
skipped. The
+ * alternatives of a `|` outside any group are tried one after another, so
each is counted on its
+ * own and the largest count is returned. These quantifiers are not counted:
+ * - a possessive one, which never backtracks;
+ * - one that ends the regex or an alternative of it, possibly inside
groups that end there too,
+ * which a failed match never backtracks into;
+ * - `?` and a `{n,m}` that spans fewer than 10, which add a small factor.
The choices of such
+ * quantifiers multiply, so they count as one quantifier each time their
product reaches 100.
+ * The count is `Int.MaxValue` for a repeated group that holds a quantifier
that can backtrack or
+ * a `|`, for a regex in COMMENTS mode, and for a `[` that is never closed,
since the scan cannot
+ * follow them.
+ *
+ * A streaming query pins `SQLConf.REGEX_PATTERN_COST_ESTIMATE_VERSION` in
its checkpoint, since
+ * a pattern that turns cheap can move a filter below a stateful operator.
So a change that makes
+ * any pattern cheaper needs a new version of the estimate, and the old one
has to be kept.
+ */
+ private def scanQuantifiers(regex: String): (Int, Boolean) = {
+ // The largest count of the alternatives before the current one.
+ var maxCount = 0
+ var count = 0
+ var smallRanges = 1
+ var classDepth = 0
+ var topLevelAlternation = false
+ // Whether each open group holds a choice that a repetition of it would
multiply, innermost
+ // first.
+ var groups = List.empty[Boolean]
+ var i = 0
+ def markGroup(): Unit = groups = groups match {
+ case _ :: outer => true :: outer
+ case Nil => Nil
+ }
+ // Whether nothing but the closing `)` of open groups comes between `j`
and the end of the
+ // regex or a `|` outside any group.
+ def endsAlternative(j: Int): Boolean = {
+ var end = j
+ var depth = groups.length
+ while (end < regex.length && depth > 0 && regex.charAt(end) == ')') {
+ end += 1
+ depth -= 1
+ }
+ end == regex.length || (depth == 0 && regex.charAt(end) == '|')
+ }
+ // Handles the quantifier that ends at `i`, which repeats its operand
`min` to `max` times,
+ // and moves `i` past its modifier.
+ def quantifier(min: Int, max: Int): Unit = {
+ val possessive = i + 1 < regex.length && regex.charAt(i + 1) == '+'
Review Comment:
Done as you suggested.
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryManagerSuite.scala:
##########
@@ -168,6 +169,36 @@ class StreamingQueryManagerSuite extends StreamTest {
}
}
+ test("SPARK-60632: regex pattern cost estimate version 0 for old
checkpoints") {
+ withTempDir { checkpointDir =>
+ val input = MemoryStream[String]
+ // `NOT` caches whether its child is expensive, and both queries share
this plan.
+ val result = input.toDF().select($"value", !$"value".rlike("^1") as
"p").where($"p")
+
+ def run(data: String): Boolean = {
+ val query = result.writeStream
+ .format("noop")
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ .start()
+ try {
+ input.addData(data)
+ query.processAllAvailable()
+ // Whether the filter stays above the projection.
+
query.asInstanceOf[StreamingQueryWrapper].streamingQuery.lastExecution.optimizedPlan
+ .exists {
+ case Filter(_, _: Project) => true
+ case _ => false
+ }
+ } finally {
+ query.stop()
+ }
+ }
+
+ assert(!run("a"))
+ rewriteOffsetLogConf(checkpointDir)(_ -
SQLConf.REGEX_PATTERN_COST_ESTIMATE_VERSION.key)
+ assert(run("b"))
+ }
+ }
Review Comment:
Done as you suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
