MaxGekk commented on code in PR #59573:
URL: https://github.com/apache/spark/pull/59573#discussion_r4333339544
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -2814,4 +2815,312 @@ 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`,
`(a > 0 OR b < 0)`
+ // and `coalesce(b, cast(s AS INT)) < 0`. The rows can be
Review Comment:
Right, the first row's file was skipped by its statistics in a multi-core
session. Added `SET spark.sql.parquet.filterPushdown=false;` after the CTAS in
the PR description, with a one-line comment on why.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -137,47 +137,182 @@ trait GeneratePredicateHelper extends PredicateHelper {
condition: Expression,
inputAttrs: Seq[Attribute],
inputExprCode: Seq[ExprCode]): String = {
- val (notNullPreds, otherPreds) =
splitConjunctivePredicates(condition).partition {
- case IsNotNull(a) => isNullIntolerant(a) &&
a.references.subsetOf(AttributeSet(inputAttrs))
- case _ => false
- }
+ val (notNullPreds, otherPreds, precedingNotNulls) =
+ splitNotNullPreds(condition, AttributeSet(inputAttrs))
val nonNullAttrExprIds =
notNullPreds.flatMap(_.references).distinct.map(_.exprId)
val outputAttrs = outputWithNullability(inputAttrs, nonNullAttrExprIds)
generatePredicateCode(
- ctx, inputAttrs, inputExprCode, outputAttrs, notNullPreds, otherPreds,
+ ctx, inputAttrs, inputExprCode, notNullPreds, otherPreds,
+ otherPreds.map(BindReferences.bindReference(_, outputAttrs)),
precedingNotNulls,
nonNullAttrExprIds)
}
+ /**
+ * Splits `condition` into the `IsNotNull` checks on null-intolerant
expressions over
+ * `inputAttrs`, the other conjuncts, and for each of the other conjuncts
the number of the
+ * checks that precede it in `condition`. The conjuncts keep their order
within each group, so
+ * the checks before the k-th other conjunct are the first
`precedingNotNulls(k)` checks.
+ */
+ protected def splitNotNullPreds(
+ condition: Expression,
+ inputAttrs: AttributeSet): (Seq[Expression], Seq[Expression], Seq[Int])
= {
+ val notNullPreds = mutable.ArrayBuffer.empty[Expression]
+ val otherPreds = mutable.ArrayBuffer.empty[Expression]
+ val precedingNotNulls = mutable.ArrayBuffer.empty[Int]
+ splitConjunctivePredicates(condition).foreach {
+ case p @ IsNotNull(a) if isNullIntolerant(a) &&
a.references.subsetOf(inputAttrs) =>
+ notNullPreds += p
+ case p =>
+ otherPreds += p
+ precedingNotNulls += notNullPreds.length
+ }
+ (notNullPreds.toSeq, otherPreds.toSeq, precedingNotNulls.toSeq)
+ }
+
+ /**
+ * Whether evaluating `e` on a row can be observed other than through its
value: it can raise
+ * an error, is nondeterministic or may have side effects, such as a UDF.
Such an expression
+ * has to be evaluated on the rows, and in the order, that `And.eval`
evaluates it on.
+ */
+ protected def mayRaise(e: Expression): Boolean =
!ExprUtils.canEvaluateUnconditionally(e)
Review Comment:
Updated the bullet in the PR description: the trait gets six helpers,
`splitNotNullPreds`, `mayRaise`, `nullConjunctFlag`, `conjunctCheck`,
`withNullConjunctFlag` and `precedingNotNullChecks`, each with a short
description.
--
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]
