ulysses-you commented on code in PR #57986:
URL: https://github.com/apache/spark/pull/57986#discussion_r3780586185
##########
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:
I do not think we should concern that, otherwise we can not optimize any
ordering related plan, such as eliminate sort, a smj -> bhj/shj when the join
is the child of a containing floating-point arithmetic Agg, also the result of
sort-based agg is different with hash-based. So if people require an accurate
result, they should use decimal type.
--
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]
