Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/18540#discussion_r125602833
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowExec.scala
---
@@ -109,46 +109,54 @@ case class WindowExec(
*
* This method uses Code Generation. It can only be used on the executor
side.
*
- * @param frameType to evaluate. This can either be Row or Range based.
- * @param offset with respect to the row.
+ * @param frame to evaluate. This can either be a Row or Range frame.
+ * @param bound with respect to the row.
* @return a bound ordering object.
*/
- private[this] def createBoundOrdering(frameType: FrameType, offset:
Int): BoundOrdering = {
- frameType match {
- case RangeFrame =>
- val (exprs, current, bound) = if (offset == 0) {
- // Use the entire order expression when the offset is 0.
- val exprs = orderSpec.map(_.child)
- val buildProjection = () => newMutableProjection(exprs,
child.output)
- (orderSpec, buildProjection(), buildProjection())
- } else if (orderSpec.size == 1) {
- // Use only the first order expression when the offset is
non-null.
- val sortExpr = orderSpec.head
- val expr = sortExpr.child
- // Create the projection which returns the current 'value'.
- val current = newMutableProjection(expr :: Nil, child.output)
- // Flip the sign of the offset when processing the order is
descending
- val boundOffset = sortExpr.direction match {
- case Descending => -offset
- case Ascending => offset
- }
- // Create the projection which returns the current 'value'
modified by adding the offset.
- val boundExpr = Add(expr, Cast(Literal.create(boundOffset,
IntegerType), expr.dataType))
- val bound = newMutableProjection(boundExpr :: Nil, child.output)
- (sortExpr :: Nil, current, bound)
- } else {
- sys.error("Non-Zero range offsets are not supported for windows
" +
- "with multiple order expressions.")
+ private[this] def createBoundOrdering(frame: FrameType, bound: AnyRef):
BoundOrdering = {
+ (frame, bound) match {
+ case (RowFrame, CurrentRow) =>
+ RowBoundOrdering(0)
+
+ case (RowFrame, IntegerLiteral(offset)) =>
+ RowBoundOrdering(offset)
+
+ case (RangeFrame, CurrentRow) =>
+ val ordering = newOrdering(orderSpec, child.output)
+ RangeBoundOrdering(ordering, IdentityProjection,
IdentityProjection)
+
+ case (RangeFrame, offset: Expression) if orderSpec.size == 1 =>
+ // Use only the first order expression when the offset is non-null.
+ val sortExpr = orderSpec.head
+ val expr = sortExpr.child
+
+ // Create the projection which returns the current 'value'.
+ val current = newMutableProjection(expr :: Nil, child.output)
+
+ // Flip the sign of the offset when processing the order is
descending
+ val boundOffset = sortExpr.direction match {
+ case Descending => UnaryMinus(offset)
+ case Ascending => offset
+ }
+
+ // Create the projection which returns the current 'value'
modified by adding the offset.
+ val boundExpr = (expr.dataType, boundOffset.dataType) match {
+ case (DateType, IntegerType) => DateAdd(expr, boundOffset)
--- End diff --
Both `DateType` and `TimestampType` expressions are going to need a time
zone. I was wondering if we can use a `GMT` because these are just offset
calculation? cc @ueshin
If we can't then we either need to thread through the session local
timezone, or it might be easier to put the entire offset calculation in the
frame.
---
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
