Skip to content

[SPARK-24168][SQL] WindowExec should not access SQLConf at executor side #21225

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
cloud-fan wants to merge 2 commits into from
Closed

[SPARK-24168][SQL] WindowExec should not access SQLConf at executor side #21225

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c5ab777
WindowExec should not access SQLConf at executor side
cloud-fan May 3, 2018
e4151a3
address comment
cloud-fan May 3, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,9 +112,11 @@ case class WindowExec(
*
* @param frame to evaluate. This can either be a Row or Range frame.
* @param bound with respect to the row.
Copy link
Contributor

@jiangxb1987 jiangxb1987 May 3, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: add param timeZone.

* @param timeZone the session local timezone for time related calculations.
* @return a bound ordering object.
*/
private[this] def createBoundOrdering(frame: FrameType, bound: Expression): BoundOrdering = {
private[this] def createBoundOrdering(
frame: FrameType, bound: Expression, timeZone: String): BoundOrdering = {
(frame, bound) match {
case (RowFrame, CurrentRow) =>
RowBoundOrdering(0)
Expand Down Expand Up @@ -144,7 +146,7 @@ case class WindowExec(
val boundExpr = (expr.dataType, boundOffset.dataType) match {
case (DateType, IntegerType) => DateAdd(expr, boundOffset)
case (TimestampType, CalendarIntervalType) =>
TimeAdd(expr, boundOffset, Some(conf.sessionLocalTimeZone))
TimeAdd(expr, boundOffset, Some(timeZone))
case (a, b) if a== b => Add(expr, boundOffset)
}
val bound = newMutableProjection(boundExpr :: Nil, child.output)
Expand Down Expand Up @@ -197,6 +199,7 @@ case class WindowExec(

// Map the groups to a (unbound) expression and frame factory pair.
var numExpressions = 0
val timeZone = conf.sessionLocalTimeZone
Copy link
Contributor

@hvanhovell hvanhovell May 3, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

NIT: should we also touch up the use of sqlContext.conf in doExecute()?

Copy link
Contributor Author

@cloud-fan cloud-fan May 3, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

that one is OK because it's not in a closure and passed to executors.

framedFunctions.toSeq.map {
case (key, (expressions, functionSeq)) =>
val ordinal = numExpressions
Expand Down Expand Up @@ -237,7 +240,7 @@ case class WindowExec(
new UnboundedPrecedingWindowFunctionFrame(
target,
processor,
createBoundOrdering(frameType, upper))
createBoundOrdering(frameType, upper, timeZone))
}

// Shrinking Frame.
Expand All @@ -246,7 +249,7 @@ case class WindowExec(
new UnboundedFollowingWindowFunctionFrame(
target,
processor,
createBoundOrdering(frameType, lower))
createBoundOrdering(frameType, lower, timeZone))
}

// Moving Frame.
Expand All @@ -255,8 +258,8 @@ case class WindowExec(
new SlidingWindowFunctionFrame(
target,
processor,
createBoundOrdering(frameType, lower),
createBoundOrdering(frameType, upper))
createBoundOrdering(frameType, lower, timeZone),
createBoundOrdering(frameType, upper, timeZone))
}
}

Expand Down