n-young commented on code in PR #60076:
URL: https://github.com/apache/spark/pull/60076#discussion_r4486142550
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -3860,6 +3866,49 @@ class Analyzer(
Project(windowOps.output ++ newExpressionsWithWindowFunctions, windowOps)
} // end of addWindow
+ /**
+ * Builds nested window dependencies above HAVING so it filters groups
before every window.
+ * Grouping helpers stay below the projection that separates HAVING from
window scope.
+ */
+ private def addWindowsWithHaving(
Review Comment:
Synced this branch onto #60090 without changing its implementation. That
prerequisite owns recursive nested-window and generator placement, controlled
by its independent legacy config. This PR only adds grouping-scope helpers and
the grouping-analytics route needed to preserve their scope. #60090 should land
first.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala:
##########
@@ -473,6 +485,104 @@ trait ColumnResolutionHelper extends Logging with
DataTypeErrorsBase {
includeLastResort = includeLastResort)
}
+ /** Hides helper grouping outputs when missing-column resolution searches an
Aggregate. */
+ private def resolveOutputWithoutHavingHelpers(
+ plan: LogicalPlan,
+ nameParts: Seq[String]): Option[NamedExpression] = {
+ if
(conf.getConf(SQLConf.HAVING_WITH_WINDOW_RESOLVES_GROUPING_COLUMN_FIRST)) {
+ resolveWithoutHavingHelpers(nameParts, plan.output,
plan.resolve(nameParts, conf.resolver))
+ } else {
+ plan.resolve(nameParts, conf.resolver)
+ }
+ }
+
+ /** Keeps helper grouping outputs out of ordinary SELECT-alias resolution. */
+ private def resolveChildrenWithHavingScope(
+ plan: LogicalPlan,
+ nameParts: Seq[String]): Option[NamedExpression] = {
+ val hasHavingScope =
+
plan.getTagValue(ColumnResolutionHelper.HAVING_GROUPING_SCOPE_TAG).contains(true)
||
+ (plan match {
+ // QUALIFY searches below the projection that trims HAVING helper
outputs.
+ case project: Project => project.child.getTagValue(
+ ColumnResolutionHelper.HAVING_GROUPING_SCOPE_TAG).contains(true)
+ case _ => false
+ })
+ if (hasHavingScope) {
+ resolveWithoutHavingHelpers(
+ nameParts, plan.children.flatMap(_.output),
plan.resolveChildren(nameParts, conf.resolver))
+ } else {
+ plan.resolveChildren(nameParts, conf.resolver)
+ }
+ }
+
+ /** Keeps ordinary lookup indexed and hides HAVING helpers in resolved
results and errors. */
+ private def resolveWithoutHavingHelpers(
+ nameParts: Seq[String],
+ output: => Seq[Attribute],
+ resolve: => Option[NamedExpression]): Option[NamedExpression] = {
+ lazy val allOutput = output
+ lazy val visibleOutput = allOutput.filterNot(
+
_.metadata.contains(ColumnResolutionHelper.HAVING_GROUPING_COLUMN_METADATA_KEY))
+ val resolved = try {
+ resolve
+ } catch {
+ // Helpers can cause ambiguity or a nested-field error before lookup
returns a candidate.
+ case _: AnalysisException if visibleOutput.length != allOutput.length =>
+ return LocalRelation(visibleOutput).resolve(nameParts, conf.resolver)
+ }
+ val mayReferenceHelpers = resolved.exists { expression =>
+ val references = expression.references
+ references.exists(
+
_.metadata.contains(ColumnResolutionHelper.HAVING_GROUPING_COLUMN_METADATA_KEY))
||
+ // NULL field extraction can discard the helper reference during
resolution.
+ (nameParts.length > 1 && references.isEmpty &&
+ visibleOutput.length != allOutput.length)
+ }
+ if (mayReferenceHelpers) {
+ LocalRelation(visibleOutput).resolve(nameParts, conf.resolver)
+ } else {
+ resolved
+ }
+ }
+
+ /**
+ * Resolves complete grouping expressions before SELECT aliases in extracted
HAVING filters.
+ * Resolve against the full input to retain ambiguity, and only accept
actual grouping
+ * expressions so ungrouped inputs cannot mask aliases or influence their
type coercion.
+ */
+ private def resolveHavingGroupingExpressions(
+ expression: Expression,
+ plan: LogicalPlan,
+ wrap: Expression => Expression = identity): Expression = {
+ if
(!plan.getTagValue(ColumnResolutionHelper.HAVING_GROUPING_SCOPE_TAG).contains(true))
{
+ return expression
+ }
+ val aggregate = plan.collectFirst { case aggregate: Aggregate => aggregate
}
+ aggregate.map { aggregate =>
+ // Leave lambda parameters and captures alone until the lambda has been
bound.
+ def resolveGrouping(expression: Expression): Expression = expression
match {
+ case _ if expression.resolved => expression
+ case lambda: LambdaFunction if !lambda.bound => lambda
+ case _ =>
+ val candidate = resolveExpressionByPlanOutput(expression,
aggregate.child) match {
+ case Alias(child, _) => child
+ case other => other
+ }
+ val grouping = if (candidate.resolved &&
+
aggregate.groupingExpressions.exists(_.semanticEquals(candidate))) {
+ // Keep inputs until aggregate extraction. UpdateOuterReferences
then maps
+ // complete correlated grouping expressions to their real
aggregate outputs.
+ Some(candidate)
Review Comment:
Extended SQL-function deferral for tagged HAVING filters whose references
are not yet aggregate outputs. Aggregate resolution now completes their
remapping before UDF inlining. Added small VALUES cases with a SQL UDF in
HAVING and in SELECT, explicitly under both grouping-scope values.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -4838,7 +4889,14 @@ object UpdateOuterReferences extends Rule[LogicalPlan] {
// Collect the aliases from output of aggregate.
val outerAliases = a.aggregateExpressions collect { case a: Alias
=> a }
// Update the subquery plan to record the OuterReference to point
to outer query plan.
- s.withNewPlan(updateOuterReferenceInSubquery(s.plan, outerAliases))
+ val newPlan = updateOuterReferenceInSubquery(s.plan, outerAliases)
+ val newSubquery = s.withNewPlan(newPlan)
+ if
(f.getTagValue(ColumnResolutionHelper.HAVING_GROUPING_SCOPE_TAG).contains(true))
{
Review Comment:
Moved correlated-reference finalization into aggregate resolution, after
extra aggregates have been pushed. It updates the subquery plan and recomputes
outer attributes there; the late rule no longer depends on the HAVING tag.
Added computed-key cases with QUALIFY and with a SQL UDF.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala:
##########
@@ -473,6 +485,104 @@ trait ColumnResolutionHelper extends Logging with
DataTypeErrorsBase {
includeLastResort = includeLastResort)
}
+ /** Hides helper grouping outputs when missing-column resolution searches an
Aggregate. */
+ private def resolveOutputWithoutHavingHelpers(
+ plan: LogicalPlan,
+ nameParts: Seq[String]): Option[NamedExpression] = {
+ if
(conf.getConf(SQLConf.HAVING_WITH_WINDOW_RESOLVES_GROUPING_COLUMN_FIRST)) {
+ resolveWithoutHavingHelpers(nameParts, plan.output,
plan.resolve(nameParts, conf.resolver))
+ } else {
+ plan.resolve(nameParts, conf.resolver)
+ }
+ }
+
+ /** Keeps helper grouping outputs out of ordinary SELECT-alias resolution. */
+ private def resolveChildrenWithHavingScope(
+ plan: LogicalPlan,
+ nameParts: Seq[String]): Option[NamedExpression] = {
+ val hasHavingScope =
+
plan.getTagValue(ColumnResolutionHelper.HAVING_GROUPING_SCOPE_TAG).contains(true)
||
+ (plan match {
+ // QUALIFY searches below the projection that trims HAVING helper
outputs.
+ case project: Project => project.child.getTagValue(
+ ColumnResolutionHelper.HAVING_GROUPING_SCOPE_TAG).contains(true)
+ case _ => false
+ })
+ if (hasHavingScope) {
+ resolveWithoutHavingHelpers(
+ nameParts, plan.children.flatMap(_.output),
plan.resolveChildren(nameParts, conf.resolver))
+ } else {
+ plan.resolveChildren(nameParts, conf.resolver)
+ }
+ }
+
+ /** Keeps ordinary lookup indexed and hides HAVING helpers in resolved
results and errors. */
+ private def resolveWithoutHavingHelpers(
+ nameParts: Seq[String],
+ output: => Seq[Attribute],
+ resolve: => Option[NamedExpression]): Option[NamedExpression] = {
+ lazy val allOutput = output
+ lazy val visibleOutput = allOutput.filterNot(
+
_.metadata.contains(ColumnResolutionHelper.HAVING_GROUPING_COLUMN_METADATA_KEY))
+ val resolved = try {
+ resolve
+ } catch {
+ // Helpers can cause ambiguity or a nested-field error before lookup
returns a candidate.
+ case _: AnalysisException if visibleOutput.length != allOutput.length =>
+ return LocalRelation(visibleOutput).resolve(nameParts, conf.resolver)
+ }
+ val mayReferenceHelpers = resolved.exists { expression =>
+ val references = expression.references
+ references.exists(
+
_.metadata.contains(ColumnResolutionHelper.HAVING_GROUPING_COLUMN_METADATA_KEY))
||
+ // NULL field extraction can discard the helper reference during
resolution.
+ (nameParts.length > 1 && references.isEmpty &&
+ visibleOutput.length != allOutput.length)
+ }
+ if (mayReferenceHelpers) {
+ LocalRelation(visibleOutput).resolve(nameParts, conf.resolver)
+ } else {
+ resolved
+ }
+ }
+
+ /**
+ * Resolves complete grouping expressions before SELECT aliases in extracted
HAVING filters.
+ * Resolve against the full input to retain ambiguity, and only accept
actual grouping
+ * expressions so ungrouped inputs cannot mask aliases or influence their
type coercion.
+ */
+ private def resolveHavingGroupingExpressions(
+ expression: Expression,
+ plan: LogicalPlan,
+ wrap: Expression => Expression = identity): Expression = {
+ if
(!plan.getTagValue(ColumnResolutionHelper.HAVING_GROUPING_SCOPE_TAG).contains(true))
{
+ return expression
+ }
+ val aggregate = plan.collectFirst { case aggregate: Aggregate => aggregate
}
+ aggregate.map { aggregate =>
+ // Leave lambda parameters and captures alone until the lambda has been
bound.
+ def resolveGrouping(expression: Expression): Expression = expression
match {
+ case _ if expression.resolved => expression
+ case lambda: LambdaFunction if !lambda.bound => lambda
+ case _ =>
+ val candidate = resolveExpressionByPlanOutput(expression,
aggregate.child) match {
+ case Alias(child, _) => child
+ case other => other
+ }
+ val grouping = if (candidate.resolved &&
Review Comment:
Added a detached bounded normalization probe for stock lower/lcase,
upper/ucase, substr/substring, arithmetic and cast keys. It accepts only
complete semantic grouping matches and discards tentative bindings and casts on
a miss. The probe checks builtin ownership and never invokes custom builders or
a nested Analyzer. Arbitrary UDF and unsupported-function grouping keys retain
the existing fallback; this is intentionally not general function resolution.
Golden controls cover lower, substr, decimal coercion, ANSI on/off and alias
fallback.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -3883,12 +3932,14 @@ class Analyzer(
throw
QueryCompilationErrors.lateralColumnAliasInAggWithWindowAndHavingUnsupportedError(
lcaRef.nameParts)
})
- val (windowExpressions, aggregateExpressions) =
extract(aggregateExprs, groupingExprs)
- // Create an Aggregate operator to evaluate aggregation functions.
- val withAggregate = Aggregate(groupingExprs, aggregateExpressions,
child)
- // Add a Filter operator for conditions in the Having clause.
- val withFilter = Filter(condition, withAggregate)
- val withWindow = addWindow(windowExpressions, withFilter)
+ val withWindow = if (
+
conf.getConf(SQLConf.HAVING_WITH_WINDOW_RESOLVES_GROUPING_COLUMN_FIRST)) {
+ addWindowsWithHaving(groupingExprs, aggregateExprs, condition, child)
Review Comment:
With grouping scope enabled and the independent legacy placement switch
disabled, grouping analytics retains UnresolvedHaving above its expanded
Aggregate so window extraction owns placement. The Expand carries original
input/key mapping: scalar grouping references use nullable expanded keys, while
aggregate arguments retain original input rows. Added GROUPING SETS, ROLLUP,
CUBE and nested-window controls under both values, including computed keys and
grand totals. Generator placement is inherited unchanged from #60090, so its
corrected result applies with either grouping-scope value.
##########
sql/core/src/test/resources/sql-tests/inputs/having.sql:
##########
@@ -127,3 +127,165 @@ SELECT col2 FROM hav_outer GROUP BY col2 HAVING (SELECT
max(col1)) > 1;
SELECT col2 FROM hav_outer GROUP BY col2 HAVING (SELECT col1);
DROP VIEW hav_outer;
+
+-- HAVING keeps grouping scope below a window; aggregate aliases keep their
types.
+SET
spark.sql.analyzer.havingWithWindowResolvesGroupingColumnFirst.enabled=true;
Review Comment:
Added explicit enabled/disabled grouping-set cases with unselected
grouping_id(), explicit grouping_id(k), grouping(k), aggregate predicates, and
ORDER BY dependencies. Enabled cases must produce the expected rows. Disabled
cases preserve existing results or legacy analysis errors rather than treating
those errors as enabled success.
--
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]
