zml1206 commented on code in PR #57986:
URL: https://github.com/apache/spark/pull/57986#discussion_r3780654296
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1761,9 +1761,38 @@ object CollapseWindow extends Rule[LogicalPlan] {
s1.zip(s2).forall(e => e._1.semanticEquals(e._2))
}
+ /**
+ * Returns true if the given window expression can be evaluated under any
ordering of the rows
+ * within a partition without changing the result, so that it can be merged
into another window
+ * with a different (non-empty) order spec.
+ *
+ * The frame determines whether the ordering matters. When the frame is the
whole partition
+ * (`UNBOUNDED PRECEDING` to `UNBOUNDED FOLLOWING`), it always covers all
the rows of the
+ * partition regardless of the ordering, so the ordering does not affect the
result: aggregates
+ * such as `count` or `sum` give the same value under any ordering, and
functions whose result
+ * does depend on the row order, such as `collect_list` or `first`, are
non-deterministic when
+ * the order spec is empty, so evaluating them under any ordering yields a
valid result. On the
+ * other hand, a bounded frame (e.g. `ROWS BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW`) is
+ * order-sensitive: which rows are in the frame depends on the ordering, so
even `count` or
+ * `sum` would change value, and such a window must not be merged.
+ */
+ private def orderInsensitive(windowExpression: NamedExpression): Boolean =
+ windowExpression match {
+ case Alias(WindowExpression(_, WindowSpecDefinition(_, _,
+ SpecifiedWindowFrame(_, UnboundedPreceding, UnboundedFollowing))),
_) => true
+ case _ => false
+ }
+
private def windowsCompatible(w1: Window, w2: Window): Boolean = {
specCompatible(w1.partitionSpec, w2.partitionSpec) &&
- specCompatible(w1.orderSpec, w2.orderSpec) &&
+ // The order specs can differ when one of them is empty, as long as the
window expressions
+ // of the window with the empty order spec are insensitive to the row
order. In that case,
+ // they can be evaluated under the non-empty order spec of the other
window.
+ (specCompatible(w1.orderSpec, w2.orderSpec) ||
+ (w1.orderSpec.isEmpty && w2.orderSpec.nonEmpty &&
+ w1.windowExpressions.forall(orderInsensitive)) ||
+ (w2.orderSpec.isEmpty && w1.orderSpec.nonEmpty &&
+ w2.windowExpressions.forall(orderInsensitive))) &&
Review Comment:
Thanks, that makes sense.
Could we adjust the comment of `orderInsensitive`? The statement that sum
gives the same value under any ordering is technically incorrect for
floating-point inputs, and orderInsensitive is also slightly misleading for
functions such as first and collect_list, whose values may change but remain
valid under their nondeterministic contract.
--
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]
