Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/6524#discussion_r31486948
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
---
@@ -705,36 +708,77 @@ class Analyzer(
}.asInstanceOf[NamedExpression]
}
- (newWindowExpressions, regularExpressions ++ extractedExprBuffer)
+ (newExpressionsWithWindowFunctions, regularExpressions ++
extractedExprBuffer)
}
/**
* Adds operators for Window Expressions. Every Window operator
handles a single Window Spec.
*/
- def addWindow(windowExpressions: Seq[NamedExpression], child:
LogicalPlan): LogicalPlan = {
- // First, we group window expressions based on their Window Spec.
- val groupedWindowExpression = windowExpressions.groupBy { expr =>
- val windowSpec = expr.collectFirst {
+ def addWindow(
+ expressionsWithWindowFunctions: Seq[NamedExpression],
+ child: LogicalPlan): LogicalPlan = {
+ // First, we need to extract all WindowExpressions from
expressionsWithWindowFunctions
+ // and put those extracted WindowExpressions to
extractedWindowExprBuffer.
+ // This step is needed because it is possible that an expression
contains multiple
+ // WindowExpressions with different Window Specs.
+ // After extracting WindowExpressions, we need to construct a
project list to generate
+ // expressionsWithWindowFunctions based on extractedWindowExprBuffer.
+ // For example, for "sum(a) over (...) / sum(b) over (...)", we will
first extract
+ // "sum(a) over (...)" and "sum(b) over (...)" out, and assign
"_we0" as the alias to
+ // "sum(a) over (...)" and "_we1" as the alias to "sum(b) over
(...)".
+ // Then, the projectList will be [_we0/_we1].
+ val extractedWindowExprBuffer = new ArrayBuffer[NamedExpression]()
+ val newExpressionsWithWindowFunctions =
expressionsWithWindowFunctions.map {
+ // We need to use transformDown because we want to trigger
+ // "case alias @ Alias(window: WindowExpression, _)" first.
+ _.transformDown {
+ case alias @ Alias(window: WindowExpression, _) =>
+ // If a WindowExpression has an assigned alias, just use it.
+ extractedWindowExprBuffer += alias
+ alias.toAttribute
+ case window: WindowExpression =>
+ // If there is no alias assigned to the WindowExpressions. We
create an
+ // internal column.
+ val withName = Alias(window,
s"_we${extractedWindowExprBuffer.length}")()
+ extractedWindowExprBuffer += withName
+ withName.toAttribute
+ }.asInstanceOf[NamedExpression]
+ }
+
+ // Second, we group extractedWindowExprBuffer based on their Window
Spec.
+ val groupedWindowExpressions = extractedWindowExprBuffer.groupBy {
expr =>
+ val distinctWindowSpec = expr.collect {
case window: WindowExpression => window.windowSpec
+ }.distinct
+
+ // We do a final check and see if we only have a single Window
Spec defined in an
+ // expressions.
+ if (distinctWindowSpec.length == 0 ) {
+ failAnalysis(s"$expr does not have any WindowExpression.")
+ } else if (distinctWindowSpec.length > 1) {
--- End diff --
add inline comment explaining why this is not possible
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
