Github user nsyca commented on a diff in the pull request:

    https://github.com/apache/spark/pull/16572#discussion_r95981409
  
    --- Diff: 
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
 ---
    @@ -117,66 +117,72 @@ trait CheckAnalysis extends PredicateHelper {
                     failAnalysis(s"Window specification $s is not valid 
because $m")
                   case None => w
                 }
    -          case s @ ScalarSubquery(query, conditions, _)
    +
    +          case e @ PredicateSubquery(query, _, _, _) =>
    +            checkAnalysis(query)
    +            e
    +
    +          case s @ ScalarSubquery(query, conditions, _) =>
                 // If no correlation, the output must be exactly one column
    -            if (conditions.isEmpty && query.output.size != 1) =>
    +            if (conditions.isEmpty && query.output.size != 1) {
                   failAnalysis(
                     s"Scalar subquery must return only one column, but got 
${query.output.size}")
    -
    -          case s @ ScalarSubquery(query, conditions, _) if 
conditions.nonEmpty =>
    -
    -            // Collect the columns from the subquery for further checking.
    -            var subqueryColumns = 
conditions.flatMap(_.references).filter(query.output.contains)
    -
    -            def checkAggregate(agg: Aggregate): Unit = {
    -              // Make sure correlated scalar subqueries contain one row 
for every outer row by
    -              // enforcing that they are aggregates which contain exactly 
one aggregate expressions.
    -              // The analyzer has already checked that subquery contained 
only one output column,
    -              // and added all the grouping expressions to the aggregate.
    -              val aggregates = agg.expressions.flatMap(_.collect {
    -                case a: AggregateExpression => a
    -              })
    -              if (aggregates.isEmpty) {
    -                failAnalysis("The output of a correlated scalar subquery 
must be aggregated")
    -              }
    -
    -              // SPARK-18504/SPARK-18814: Block cases where GROUP BY 
columns
    -              // are not part of the correlated columns.
    -              val groupByCols = 
AttributeSet(agg.groupingExpressions.flatMap(_.references))
    -              val correlatedCols = AttributeSet(subqueryColumns)
    -              val invalidCols = groupByCols -- correlatedCols
    -              // GROUP BY columns must be a subset of columns in the 
predicates
    -              if (invalidCols.nonEmpty) {
    -                failAnalysis(
    -                  "A GROUP BY clause in a scalar correlated subquery " +
    -                    "cannot contain non-correlated columns: " +
    -                    invalidCols.mkString(","))
    -              }
                 }
    +            else if (conditions.nonEmpty) {
    +              // Collect the columns from the subquery for further 
checking.
    +              var subqueryColumns = 
conditions.flatMap(_.references).filter(query.output.contains)
    +
    +              def checkAggregate(agg: Aggregate): Unit = {
    +                // Make sure correlated scalar subqueries contain one row 
for every outer row by
    +                // enforcing that they are aggregates containing exactly 
one aggregate expression.
    +                // The analyzer has already checked that subquery 
contained only one output column,
    +                // and added all the grouping expressions to the aggregate.
    +                val aggregates = agg.expressions.flatMap(_.collect {
    +                  case a: AggregateExpression => a
    +                })
    +                if (aggregates.isEmpty) {
    +                  failAnalysis("The output of a correlated scalar subquery 
must be aggregated")
    +                }
     
    -            // Skip subquery aliases added by the Analyzer and the 
SQLBuilder.
    -            // For projects, do the necessary mapping and skip to its 
child.
    -            def cleanQuery(p: LogicalPlan): LogicalPlan = p match {
    -              case s: SubqueryAlias => cleanQuery(s.child)
    -              case p: Project =>
    -                // SPARK-18814: Map any aliases to their 
AttributeReference children
    -                // for the checking in the Aggregate operators below this 
Project.
    -                subqueryColumns = subqueryColumns.map {
    -                  xs => p.projectList.collectFirst {
    -                    case e @ Alias(child : AttributeReference, _) if 
e.exprId == xs.exprId =>
    -                      child
    -                  }.getOrElse(xs)
    +                // SPARK-18504/SPARK-18814: Block cases where GROUP BY 
columns
    +                // are not part of the correlated columns.
    +                val groupByCols = 
AttributeSet(agg.groupingExpressions.flatMap(_.references))
    +                val correlatedCols = AttributeSet(subqueryColumns)
    +                val invalidCols = groupByCols -- correlatedCols
    +                // GROUP BY columns must be a subset of columns in the 
predicates
    +                if (invalidCols.nonEmpty) {
    +                  failAnalysis(
    +                    "A GROUP BY clause in a scalar correlated subquery " +
    +                      "cannot contain non-correlated columns: " +
    +                      invalidCols.mkString(","))
                     }
    +              }
     
    -                cleanQuery(p.child)
    -              case child => child
    -            }
    +              // Skip subquery aliases added by the Analyzer and the 
SQLBuilder.
    +              // For projects, do the necessary mapping and skip to its 
child.
    +              def cleanQuery(p: LogicalPlan): LogicalPlan = p match {
    +                case s: SubqueryAlias => cleanQuery(s.child)
    +                case p: Project =>
    +                  // SPARK-18814: Map any aliases to their 
AttributeReference children
    +                  // for the checking in the Aggregate operators below 
this Project.
    +                  subqueryColumns = subqueryColumns.map {
    +                    xs => p.projectList.collectFirst {
    +                      case e @ Alias(child : AttributeReference, _) if 
e.exprId == xs.exprId =>
    +                        child
    +                    }.getOrElse(xs)
    +                  }
    +
    +                  cleanQuery(p.child)
    +                case child => child
    +              }
     
    -            cleanQuery(query) match {
    -              case a: Aggregate => checkAggregate(a)
    -              case Filter(_, a: Aggregate) => checkAggregate(a)
    -              case fail => failAnalysis(s"Correlated scalar subqueries 
must be Aggregated: $fail")
    +              cleanQuery(query) match {
    +                case a: Aggregate => checkAggregate(a)
    +                case Filter(_, a: Aggregate) => checkAggregate(a)
    +                case fail => failAnalysis(s"Correlated scalar subqueries 
must be Aggregated: $fail")
    +              }
                 }
    +            checkAnalysis(query)
    --- End diff --
    
    The best way to view this block of code changes is using a `diff` with 
`-b`. The main part is to call `checkAnalysis` for both PredicateSubquery and 
ScalaSubquery. 


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org

Reply via email to