Skip to content

[SPARK-20872][SQL] ShuffleExchange.nodeName should handle null coordinator #18095

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
rednaxelafx wants to merge 1 commit into from
Closed

[SPARK-20872][SQL] ShuffleExchange.nodeName should handle null coordinator #18095

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
9 changes: 6 additions & 3 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchange.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,14 +40,17 @@ case class ShuffleExchange(
child: SparkPlan,
@transient coordinator: Option[ExchangeCoordinator]) extends Exchange {

// NOTE: coordinator can be null after serialization/deserialization,
// e.g. it can be null on the Executor side

Copy link
Contributor

@jiangxb1987 jiangxb1987 May 24, 2017

Choose a reason for hiding this comment

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

nit: remove this empty line.

Copy link
Contributor Author

@rednaxelafx rednaxelafx May 24, 2017

Choose a reason for hiding this comment

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

This note is meant to be a class internal comment, and not a comment on override lazy val metrics = Map(, so I'd say leaving the added empty line here makes more sense. Would you agree?

Copy link
Contributor

@jiangxb1987 jiangxb1987 May 24, 2017

Choose a reason for hiding this comment

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

Sounds good to me.

override lazy val metrics = Map(
"dataSize" -> SQLMetrics.createSizeMetric(sparkContext, "data size"))

override def nodeName: String = {
val extraInfo = coordinator match {
case Some(exchangeCoordinator) =>
s"(coordinator id: ${System.identityHashCode(exchangeCoordinator)})"
case None => ""
case _ => ""
Copy link
Member

@sameeragarwal sameeragarwal May 24, 2017

Choose a reason for hiding this comment

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

should we adjust this pattern match condition at other places in the class below as well?

Copy link
Contributor Author

@rednaxelafx rednaxelafx May 24, 2017

Choose a reason for hiding this comment

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

I had left the other match conditions unchanged so that this change itself can be small and contained, because I'd need to verify whether or not the other match conditions should only be invoked with some valid Option[ExchangeCoordinator] value.

Anyway, I checked the two other coordinator match { ... } cases below, and both of them should still work fine to change to match on _ instead of None. But semantically both of these cases should only be invoked on the Driver side, where the coordinator shouldn't be null. I'm okay to change these two cases if reviewers suggest so.

Copy link
Contributor

@jiangxb1987 jiangxb1987 May 24, 2017

Choose a reason for hiding this comment

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

Let's do it then, that will enable future readers easier to understand the code.

}

val simpleNodeName = "Exchange"
Expand All @@ -70,7 +73,7 @@ case class ShuffleExchange(
// the plan.
coordinator match {
case Some(exchangeCoordinator) => exchangeCoordinator.registerExchange(this)
case None =>
case _ =>
}
}

Expand Down Expand Up @@ -117,7 +120,7 @@ case class ShuffleExchange(
val shuffleRDD = exchangeCoordinator.postShuffleRDD(this)
assert(shuffleRDD.partitions.length == newPartitioning.numPartitions)
shuffleRDD
case None =>
case _ =>
val shuffleDependency = prepareShuffleDependency()
preparePostShuffleRDD(shuffleDependency)
}
Expand Down