MaxGekk commented on code in PR #59573:
URL: https://github.com/apache/spark/pull/59573#discussion_r4328260173
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -2814,4 +2815,274 @@ class WholeStageCodegenSuite extends SharedSparkSession
}
}
}
+
+ // A Parquet table, so that the filter is not evaluated by the optimizer on
a local relation.
+ // On the first row `NOT (a > 0 AND b > 0)` is NULL, and no `isnotnull(a)`
is inferred from it.
+ // The second row passes every condition of the tests below, except `b < 0`.
The rows can be
+ // written to different files, so the filter is not pushed down, which would
skip the first
+ // row's file by its statistics.
+ private def withNullConjunctTable(y: Int)(f: => Unit): Unit = {
+ withTable("t") {
+ sql("CREATE TABLE t USING parquet AS " +
+ s"SELECT * FROM VALUES (NULL, 1, $y, NULL, 'x'), (0, 1, 0, 1, '1') AS
v(a, b, y, x, s)")
+ f
+ }
+ }
+
+ // A common subexpression after the NULL conjunct takes the CSE path of
FilterExec.
+ private val nullConjunctCSEQuery =
+ "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0 AND y + 1 > -1"
+
+ private def checkArithmeticOverflow(query: String): Unit = {
+ checkError(
+ exception = intercept[SparkArithmeticException] {
+ sql(query).collect()
+ },
+ condition = "ARITHMETIC_OVERFLOW",
+ parameters = Map(
+ "message" -> "overflow",
+ "alternative" -> " Use 'try_add' to tolerate overflow and return NULL
instead.",
+ "config" -> toSQLConf(SQLConf.ANSI_ENABLED.key)),
+ context = ExpectedContext(
+ fragment = "y + 1",
+ start = query.indexOf("y + 1"),
+ stop = query.indexOf("y + 1") + 4))
+ }
+
+ // Runs `f` with whole-stage codegen on and off and with the legacy behavior
on and off, and
+ // passes it whether the filter drops a row at its first NULL conjunct.
+ private def withNullConjunctConfs(otherConfs: (String, String)*)(f: Boolean
=> Unit): Unit = {
+ for (codegen <- Seq(true, false); legacy <- Seq(true, false)) {
+ withSQLConf(otherConfs ++ Seq(
+ SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "false",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString,
+ SQLConf.LEGACY_FILTER_SHORT_CIRCUIT_ON_NULL_CONJUNCT.key ->
legacy.toString): _*) {
+ withClue(s"wholeStage=$codegen, legacy=$legacy: ") {
+ f(codegen && legacy)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60315: a filter evaluates the conjuncts after a NULL one") {
+ withNullConjunctTable(Int.MaxValue) {
+ // The CSE path evaluates the common `y + 1` once, the non-CSE path at
each use. Both
+ // generate the flag that keeps the row's evaluation going after a NULL
conjunct.
+ def cseQueryCode(filterExecCSE: Boolean): String = {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
filterExecCSE.toString) {
+ val code =
codegenString(sql(nullConjunctCSEQuery).queryExecution.executedPlan)
+ assert(code.contains("nullConjunct"))
+ code
+ }
+ }
+ assert("addExact".r.findAllIn(cseQueryCode(filterExecCSE = true)).length
<
+ "addExact".r.findAllIn(cseQueryCode(filterExecCSE = false)).length)
+
+ Seq("true", "false").foreach { filterExecCSE =>
+ Seq(
+ "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0",
+ nullConjunctCSEQuery
+ ).foreach { query =>
+ withNullConjunctConfs(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
filterExecCSE) {
+ skipsAtNull =>
+ if (skipsAtNull) {
+ checkAnswer(sql(query), Row(0))
+ } else {
+ checkArithmeticOverflow(query)
+ }
+ }
+ }
+ }
+ }
+ // A NULL conjunct from OR. The second row does not pass it.
+ withNullConjunctTable(Int.MaxValue) {
+ val query = "SELECT a FROM t WHERE (a > 0 OR b < 0) AND y + 1 > 0"
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "true") { skipsAtNull
=>
+ if (skipsAtNull) {
+ checkAnswer(sql(query), Seq.empty)
+ } else {
+ checkArithmeticOverflow(query)
+ }
+ }
+ }
+ // Without an error, the CSE path drops the first row after its last
conjunct is evaluated.
+ withNullConjunctTable(0) {
+ withNullConjunctConfs(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key -> "true")
{ _ =>
+ checkAnswer(sql(nullConjunctCSEQuery), Row(0))
+ }
+ }
+ }
+
+ test("SPARK-60315: a filter evaluates the conjuncts after a NULL one in
non-ANSI mode") {
+ // Year-month interval arithmetic raises on overflow in every mode.
+ withNullConjunctTable(178956970) {
+ val condition = "NOT (a > 0 AND b > 0) AND " +
+ "make_ym_interval(y, 0) + INTERVAL '1' YEAR > INTERVAL '0' YEAR"
+ val query = s"SELECT a FROM t WHERE $condition"
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "false") { skipsAtNull
=>
+ if (skipsAtNull) {
+ checkAnswer(sql(query), Row(0))
+ } else {
+ checkError(
+ exception = intercept[SparkArithmeticException] {
+ sql(query).collect()
+ },
+ condition = "INTERVAL_ARITHMETIC_OVERFLOW.WITH_SUGGESTION",
+ parameters = Map("functionName" -> toSQLId("try_add")))
+ }
+ // A check on an expression, not on a column, that precedes the NULL
conjunct:
+ // `cast(s AS INT)` is NULL on the first row, so the check is FALSE
and drops the row
+ // before the overflowing conjunct, as with codegen off.
+ checkAnswer(
+ sql(s"SELECT a FROM t WHERE cast(s AS INT) IS NOT NULL AND
$condition"), Row(0))
+ }
+ }
+ }
+
+ test("SPARK-60315: a filter drops a row with a NULL conjunct and stops at a
FALSE one") {
+ withNullConjunctTable(Int.MaxValue) {
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "true") { _ =>
+ // The first row is dropped after its last conjunct is evaluated.
+ checkAnswer(sql("SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y >
-1"), Row(0))
+ // FALSE decides the AND, so the overflowing conjunct after it is not
evaluated.
+ checkAnswer(
+ sql("SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND b < 0 AND y + 1
> 0"),
+ Seq.empty)
+ }
+ // `b > 0` cannot be NULL after the inferred `isnotnull(b)`, so no flag
is generated.
+ val df = sql("SELECT a FROM t WHERE b > 0 AND y > -1")
+ checkAnswer(df, Seq(Row(null), Row(0)))
+
assert(!codegenString(df.queryExecution.executedPlan).contains("nullConjunct"))
Review Comment:
Good catch. Changed to
`genCode(sql(query)).exists(_.body.contains("nullConjunct"))` in f81f1c81082,
so the assertion fails if the filter is not in a whole-stage codegen subtree;
the same helper checks that the flag is present once a conjunct can raise.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -2814,4 +2815,274 @@ class WholeStageCodegenSuite extends SharedSparkSession
}
}
}
+
+ // A Parquet table, so that the filter is not evaluated by the optimizer on
a local relation.
+ // On the first row `NOT (a > 0 AND b > 0)` is NULL, and no `isnotnull(a)`
is inferred from it.
+ // The second row passes every condition of the tests below, except `b < 0`.
The rows can be
Review Comment:
Listed them in f81f1c81082: the second row passes every condition except `b
< 0`, `(a > 0 OR b < 0)` and `coalesce(b, cast(s AS INT)) < 0`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -373,29 +503,43 @@ case class FilterExec(condition: Expression, child:
SparkPlan)
// Emit an IsNotNull check, binding against child.output so the inner
expression
// is evaluated with its original nullability (the
tightened-nullability `output`
// is only appropriate inside otherPreds, where the required checks
have fired).
- // The `bound.nullable` gate below is defensive -- `IsNotNull` always
produces a
- // non-nullable Boolean, so this branch is structurally unreachable
today; it
- // mirrors the non-CSE helper's shape to keep future refactors safe.
+ // Passing `bound.nullable` to `conjunctCheck` is defensive --
`IsNotNull` always
+ // produces a non-nullable Boolean, so its NULL handling is
unreachable here; the
+ // non-CSE helper generates its IsNotNull checks the same way.
def genNotNull(pred: Expression): String = {
val bound = BindReferences.bindReference(pred, child.output)
val evaluated = evaluateRequiredVariables(child.output, input,
pred.references)
val ev = ExpressionCanonicalizer.execute(bound).genCode(ctx)
- val nullCheck = if (bound.nullable) s"${ev.isNull} || " else ""
s"""
|$evaluated
|${ev.code}
- |if (${nullCheck}!${ev.value}) continue;
+ |${conjunctCheck(ev, bound.nullable, nullFlag)}
Review Comment:
Agreed, a check must keep dropping the row before the conjuncts and
precomputes that are bound with its column non-nullable. In f81f1c81082 both
`genNotNull` helpers pass `None`, so the defensive branch is `if (isNull ||
!value) continue;` again; the generated code does not change since `IsNotNull`
is never nullable.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -451,10 +597,11 @@ case class FilterExec(condition: Expression, child:
SparkPlan)
case (p, i) if !generatedIsNotNullChecks(i) => genNotNull(p)
}.mkString("\n")
- (inputVarsEvalCode, predCode + "\n" + leftoverNotNull)
+ (inputVarsEvalCode, withNullConjunctFlag(predCode + "\n" +
leftoverNotNull, nullFlag))
} else {
("", generatePredicateCode(
- ctx, child.output, input, output, notNullPreds, otherPreds,
notNullAttributes))
+ ctx, child.output, input, notNullPreds, otherPreds, boundOtherPreds,
Review Comment:
Updated in f81f1c81082: "shared by the CSE gate in `doConsume` and both
predicate code paths, the CSE branch and `generatePredicateCode`".
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -201,56 +316,64 @@ trait GeneratePredicateHelper extends PredicateHelper {
if (idx != -1 && !generatedIsNotNullChecks(idx)) {
generatedIsNotNullChecks(idx) = true
// Use the child's output. The nullability is what the child
produced.
- genPredicate(notNullPreds(idx), inputExprCode, inputAttrs)
+ genNotNull(notNullPreds(idx))
} else if (nonNullAttrExprIds.contains(r.exprId) &&
!extraIsNotNullAttrs.contains(r)) {
Review Comment:
Right. In f81f1c81082 the synthetic branch is guarded with `idx == -1 ||
!afterNullable` (`ni == -1 || !afterNullable` in the CSE branch), so after a
nullable conjunct a synthetic check is generated only for a column that no
direct check covers, and the generated code without the flag stays identical to
master, including the redundant check on the `BETWEEN` shape, to keep the
change within this PR.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -373,29 +503,43 @@ case class FilterExec(condition: Expression, child:
SparkPlan)
// Emit an IsNotNull check, binding against child.output so the inner
expression
// is evaluated with its original nullability (the
tightened-nullability `output`
// is only appropriate inside otherPreds, where the required checks
have fired).
- // The `bound.nullable` gate below is defensive -- `IsNotNull` always
produces a
- // non-nullable Boolean, so this branch is structurally unreachable
today; it
- // mirrors the non-CSE helper's shape to keep future refactors safe.
+ // Passing `bound.nullable` to `conjunctCheck` is defensive --
`IsNotNull` always
+ // produces a non-nullable Boolean, so its NULL handling is
unreachable here; the
+ // non-CSE helper generates its IsNotNull checks the same way.
def genNotNull(pred: Expression): String = {
val bound = BindReferences.bindReference(pred, child.output)
val evaluated = evaluateRequiredVariables(child.output, input,
pred.references)
val ev = ExpressionCanonicalizer.execute(bound).genCode(ctx)
- val nullCheck = if (bound.nullable) s"${ev.isNull} || " else ""
s"""
|$evaluated
|${ev.code}
- |if (${nullCheck}!${ev.value}) continue;
+ |${conjunctCheck(ev, bound.nullable, nullFlag)}
""".stripMargin
}
val generatedIsNotNullChecks = new Array[Boolean](notNullPreds.length)
val extraIsNotNullAttrs = mutable.Set[Attribute]()
+ var nullableSeen = false
val predCode: String = {
val parts = new StringBuilder
ctx.withSubExprEliminationExprs(subExprs.states) {
otherPreds.zip(boundOtherPreds).zipWithIndex.foreach {
case ((orig, bound), idx) =>
+ val afterNullable = nullFlag.isDefined && nullableSeen
+ precedingNotNullChecks(
+ notNullPreds, precedingNotNulls, idx, afterNullable,
+ generatedIsNotNullChecks).foreach { p =>
+ // The inner expression of a check is evaluated with no CSE
state, as below.
Review Comment:
Moved into `genNotNullWithoutCSE` next to `genNotNull` in f81f1c81082, with
a comment on the reason: the check is bound against the child's output with its
original nullability, while the CSE states are computed over `boundOtherPreds`
bound against the tightened output, and a state precomputed for the check could
materialize a NULL into a shared variable, see (c). The calls stay in the same
order.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -2814,4 +2815,274 @@ class WholeStageCodegenSuite extends SharedSparkSession
}
}
}
+
+ // A Parquet table, so that the filter is not evaluated by the optimizer on
a local relation.
+ // On the first row `NOT (a > 0 AND b > 0)` is NULL, and no `isnotnull(a)`
is inferred from it.
+ // The second row passes every condition of the tests below, except `b < 0`.
The rows can be
+ // written to different files, so the filter is not pushed down, which would
skip the first
+ // row's file by its statistics.
+ private def withNullConjunctTable(y: Int)(f: => Unit): Unit = {
+ withTable("t") {
+ sql("CREATE TABLE t USING parquet AS " +
+ s"SELECT * FROM VALUES (NULL, 1, $y, NULL, 'x'), (0, 1, 0, 1, '1') AS
v(a, b, y, x, s)")
+ f
+ }
+ }
+
+ // A common subexpression after the NULL conjunct takes the CSE path of
FilterExec.
+ private val nullConjunctCSEQuery =
+ "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0 AND y + 1 > -1"
+
+ private def checkArithmeticOverflow(query: String): Unit = {
+ checkError(
+ exception = intercept[SparkArithmeticException] {
+ sql(query).collect()
+ },
+ condition = "ARITHMETIC_OVERFLOW",
+ parameters = Map(
+ "message" -> "overflow",
+ "alternative" -> " Use 'try_add' to tolerate overflow and return NULL
instead.",
+ "config" -> toSQLConf(SQLConf.ANSI_ENABLED.key)),
+ context = ExpectedContext(
+ fragment = "y + 1",
+ start = query.indexOf("y + 1"),
+ stop = query.indexOf("y + 1") + 4))
+ }
+
+ // Runs `f` with whole-stage codegen on and off and with the legacy behavior
on and off, and
+ // passes it whether the filter drops a row at its first NULL conjunct.
+ private def withNullConjunctConfs(otherConfs: (String, String)*)(f: Boolean
=> Unit): Unit = {
+ for (codegen <- Seq(true, false); legacy <- Seq(true, false)) {
+ withSQLConf(otherConfs ++ Seq(
+ SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "false",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString,
+ SQLConf.LEGACY_FILTER_SHORT_CIRCUIT_ON_NULL_CONJUNCT.key ->
legacy.toString): _*) {
+ withClue(s"wholeStage=$codegen, legacy=$legacy: ") {
+ f(codegen && legacy)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60315: a filter evaluates the conjuncts after a NULL one") {
+ withNullConjunctTable(Int.MaxValue) {
+ // The CSE path evaluates the common `y + 1` once, the non-CSE path at
each use. Both
+ // generate the flag that keeps the row's evaluation going after a NULL
conjunct.
+ def cseQueryCode(filterExecCSE: Boolean): String = {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
filterExecCSE.toString) {
+ val code =
codegenString(sql(nullConjunctCSEQuery).queryExecution.executedPlan)
+ assert(code.contains("nullConjunct"))
+ code
+ }
+ }
+ assert("addExact".r.findAllIn(cseQueryCode(filterExecCSE = true)).length
<
+ "addExact".r.findAllIn(cseQueryCode(filterExecCSE = false)).length)
+
+ Seq("true", "false").foreach { filterExecCSE =>
+ Seq(
+ "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0",
+ nullConjunctCSEQuery
+ ).foreach { query =>
+ withNullConjunctConfs(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
filterExecCSE) {
+ skipsAtNull =>
+ if (skipsAtNull) {
+ checkAnswer(sql(query), Row(0))
+ } else {
+ checkArithmeticOverflow(query)
+ }
+ }
+ }
+ }
+ }
+ // A NULL conjunct from OR. The second row does not pass it.
+ withNullConjunctTable(Int.MaxValue) {
+ val query = "SELECT a FROM t WHERE (a > 0 OR b < 0) AND y + 1 > 0"
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "true") { skipsAtNull
=>
+ if (skipsAtNull) {
+ checkAnswer(sql(query), Seq.empty)
+ } else {
+ checkArithmeticOverflow(query)
+ }
+ }
+ }
+ // Without an error, the CSE path drops the first row after its last
conjunct is evaluated.
+ withNullConjunctTable(0) {
+ withNullConjunctConfs(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key -> "true")
{ _ =>
+ checkAnswer(sql(nullConjunctCSEQuery), Row(0))
+ }
+ }
+ }
+
+ test("SPARK-60315: a filter evaluates the conjuncts after a NULL one in
non-ANSI mode") {
+ // Year-month interval arithmetic raises on overflow in every mode.
+ withNullConjunctTable(178956970) {
+ val condition = "NOT (a > 0 AND b > 0) AND " +
+ "make_ym_interval(y, 0) + INTERVAL '1' YEAR > INTERVAL '0' YEAR"
+ val query = s"SELECT a FROM t WHERE $condition"
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "false") { skipsAtNull
=>
+ if (skipsAtNull) {
+ checkAnswer(sql(query), Row(0))
+ } else {
+ checkError(
+ exception = intercept[SparkArithmeticException] {
+ sql(query).collect()
+ },
+ condition = "INTERVAL_ARITHMETIC_OVERFLOW.WITH_SUGGESTION",
+ parameters = Map("functionName" -> toSQLId("try_add")))
+ }
+ // A check on an expression, not on a column, that precedes the NULL
conjunct:
+ // `cast(s AS INT)` is NULL on the first row, so the check is FALSE
and drops the row
+ // before the overflowing conjunct, as with codegen off.
+ checkAnswer(
+ sql(s"SELECT a FROM t WHERE cast(s AS INT) IS NOT NULL AND
$condition"), Row(0))
+ }
+ }
+ }
+
+ test("SPARK-60315: a filter drops a row with a NULL conjunct and stops at a
FALSE one") {
+ withNullConjunctTable(Int.MaxValue) {
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "true") { _ =>
+ // The first row is dropped after its last conjunct is evaluated.
+ checkAnswer(sql("SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y >
-1"), Row(0))
+ // FALSE decides the AND, so the overflowing conjunct after it is not
evaluated.
+ checkAnswer(
+ sql("SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND b < 0 AND y + 1
> 0"),
+ Seq.empty)
+ }
+ // `b > 0` cannot be NULL after the inferred `isnotnull(b)`, so no flag
is generated.
+ val df = sql("SELECT a FROM t WHERE b > 0 AND y > -1")
+ checkAnswer(df, Seq(Row(null), Row(0)))
+
assert(!codegenString(df.queryExecution.executedPlan).contains("nullConjunct"))
+ }
+ }
+
+ test("SPARK-60315: a filter checks IS NOT NULL before the conjuncts it
precedes") {
+ withNullConjunctTable(Int.MaxValue) {
+ // `x` is NULL on the first row. With codegen off, a FALSE
`isnotnull(x)` drops the row
+ // before a conjunct after it in the condition is evaluated. With
codegen on, the check is
+ // generated just before the first conjunct that references `x`, or
after all of them, so
+ // after a NULL conjunct it has to be generated before the conjuncts it
precedes.
+ //
+ // The common `b * 2` takes the CSE path of FilterExec. The overflowing
`y + 1` is not
+ // common because `GeneratePredicate` (codegen off) evaluates common
subexpressions first.
+ val cseQuery = "SELECT a FROM t WHERE x IS NOT NULL AND NOT (a > 0 AND b
> 0) AND " +
+ "y + 1 > 0 AND b * 2 > 0 AND b * 2 > -1"
+ def multiplyExactCount(filterExecCSE: Boolean): Int = {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
filterExecCSE.toString) {
+ val code = codegenString(sql(cseQuery).queryExecution.executedPlan)
+ "multiplyExact".r.findAllIn(code).length
+ }
+ }
+ assert(multiplyExactCount(filterExecCSE = true) <
multiplyExactCount(filterExecCSE = false))
+
+ Seq("true", "false").foreach { filterExecCSE =>
+ withNullConjunctConfs(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
filterExecCSE) {
+ skipsAtNull =>
+ // `isnotnull(x)` is inferred from `x > 0` and precedes the
overflowing conjunct.
+ checkAnswer(
+ sql("SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0
AND x > 0"),
+ Row(0))
+ // A user-written check that no other conjunct references.
+ checkAnswer(
+ sql("SELECT a FROM t WHERE x IS NOT NULL AND NOT (a > 0 AND b >
0) AND y + 1 > 0"),
+ Row(0))
+ // The same with a common subexpression.
+ checkAnswer(sql(cseQuery), Row(0))
+ // A check written after the overflowing conjunct stays after it.
+ val checkAfter =
+ "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0 AND x
IS NOT NULL"
+ if (skipsAtNull) {
+ checkAnswer(sql(checkAfter), Row(0))
+ } else {
+ checkArithmeticOverflow(checkAfter)
+ }
+ }
+ }
+ // The aggregate FILTER clause goes through the same code generation.
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "true") { _ =>
+ checkAnswer(
+ sql("SELECT count(1) FILTER (WHERE x IS NOT NULL AND NOT (a > 0 AND
b > 0) AND " +
+ "y + 1 > 0) FROM t"),
+ Row(1))
+ }
+ }
+ }
+
+ test("SPARK-60315: a filter checks IS NOT NULL on an expression after the
conjunct it follows") {
+ withNullConjunctTable(Int.MaxValue) {
+ // `cast(s AS INT)` raises on the first row. With codegen on, a check on
an expression is
+ // generated just before the first conjunct that contains the expression
(SPARK-51356),
+ // even when it is written after that conjunct. After a NULL conjunct,
it has to stay after
+ // the conjunct, as with codegen off: `coalesce(b, cast(s AS INT)) < 0`
is FALSE on the row
+ // and drops it before the check raises, while `coalesce(b, cast(s AS
INT)) < 2` is TRUE and
+ // the check raises after it. The second row passes the latter condition.
+ //
+ // The common `b * 2` takes the CSE path of FilterExec.
+ def query(conjunct: String, cse: Boolean): String = {
+ val common = if (cse) "b * 2 > 0 AND b * 2 > -1 AND " else ""
+ s"SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND $common$conjunct AND
" +
+ "cast(s AS INT) IS NOT NULL"
+ }
+ def multiplyExactCount(filterExecCSE: Boolean): Int = {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
filterExecCSE.toString) {
+ val plan = sql(query("coalesce(b, cast(s AS INT)) < 0", cse =
true)).queryExecution
+ "multiplyExact".r.findAllIn(codegenString(plan.executedPlan)).length
+ }
+ }
+ assert(multiplyExactCount(filterExecCSE = true) <
multiplyExactCount(filterExecCSE = false))
+
+ Seq(true, false).foreach { cse =>
+ withNullConjunctConfs(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
cse.toString) {
+ skipsAtNull =>
+ checkAnswer(sql(query("coalesce(b, cast(s AS INT)) < 0", cse)),
Seq.empty)
+ val raising = query("coalesce(b, cast(s AS INT)) < 2", cse)
+ if (skipsAtNull) {
+ checkAnswer(sql(raising), Row(0))
+ } else {
+ checkError(
+ exception = intercept[SparkNumberFormatException] {
+ sql(raising).collect()
+ },
+ condition = "CAST_INVALID_INPUT",
+ parameters = Map(
+ "expression" -> "'x'",
+ "sourceType" -> "\"STRING\"",
+ "targetType" -> "\"INT\"",
+ "ansiConfig" -> toSQLConf(SQLConf.ANSI_ENABLED.key)),
+ context = ExpectedContext(
+ fragment = "cast(s AS INT)",
+ start = raising.lastIndexOf("cast(s AS INT)"),
+ stop = raising.lastIndexOf("cast(s AS INT)") + 13))
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-60315: an aggregate FILTER evaluates the conjuncts after a NULL
one") {
+ withNullConjunctTable(Int.MaxValue) {
+ val query = "SELECT count(a) FILTER (WHERE NOT (a > 0 AND b > 0) AND y +
1 > 0) FROM t"
Review Comment:
Confirmed: `AggregateCodegenSupport` emits the common subexpressions of the
update expressions before the per-function `FILTER` block, while
`AggregationIterator` wraps the update in `If(filter, ...)`, so `avg(y + 1)
FILTER (WHERE b < 0)` raises only with codegen on. Pre-existing and outside the
condition itself, so I filed SPARK-60393 and listed it under "Not fixed here".
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -186,13 +290,24 @@ trait GeneratePredicateHelper extends PredicateHelper {
// This has the property of not doing redundant IsNotNull checks and
taking better advantage of
// short-circuiting, not loading attributes until they are needed.
// This is very perf sensitive.
+ // With the NULL-conjunct flag, once a predicate that can be NULL is
generated, the IsNotNull
+ // checks that precede a later predicate in the condition are generated
before it, see
+ // `precedingNotNullChecks`, and a check on an expression that is written
after the predicate
+ // is not generated before it, see the nested checks below.
// NOTE: `FilterExec.doConsume`'s CSE branch inlines the same interleaving
so it can
Review Comment:
Agreed, the two emitters should consume one precomputed sequence of checks
and conjuncts, and SPARK-60329 and SPARK-60377 would then be one edit each.
Filed SPARK-60394 for the refactoring; I would rather land this fix first,
since the sequence depends on the placement rules settled here, and base the
two follow-ups on the refactoring.
--
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]
