Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
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
Prev Previous commit
Next Next commit
address comments
  • Loading branch information
Davies Liu committed Mar 8, 2016
commit 7df43ca78846966b0af8045b924d646d97505925
Original file line number Diff line number Diff line change
Expand Up @@ -239,9 +239,9 @@ abstract class QueryPlan[PlanType <: QueryPlan[PlanType]] extends TreeNode[PlanT
override def innerChildren: Seq[PlanType] = subqueries

/**
* Cleaned copy of this query plan.
* Canonicalized copy of this query plan.
Copy link
Contributor

Choose a reason for hiding this comment

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

@nongli, is "canonicalized" sufficiently unambiguous here or do we need to explain what this means?

*/
protected lazy val cleaned: PlanType = this
protected lazy val canonicalized: PlanType = this

/**
* Returns true when the given query plan will return the same results as this query plan.
Expand All @@ -257,8 +257,8 @@ abstract class QueryPlan[PlanType <: QueryPlan[PlanType]] extends TreeNode[PlanT
* can do better should override this function.
*/
def sameResult(plan: PlanType): Boolean = {
val cleanLeft = this.cleaned
val cleanRight = plan.cleaned
val cleanLeft = this.canonicalized
Copy link
Contributor

Choose a reason for hiding this comment

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

Should this be renamed to canonicalizedLeft?

val cleanRight = plan.canonicalized
cleanLeft.getClass == cleanRight.getClass &&
cleanLeft.children.size == cleanRight.children.size &&
cleanLeft.cleanArgs == cleanRight.cleanArgs &&
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -114,7 +114,7 @@ abstract class LogicalPlan extends QueryPlan[LogicalPlan] with Logging {
*/
def childrenResolved: Boolean = children.forall(_.resolved)

override lazy val cleaned: LogicalPlan = EliminateSubqueryAliases(this)
override lazy val canonicalized: LogicalPlan = EliminateSubqueryAliases(this)

/**
* Optionally resolves the given strings to a [[NamedExpression]] using the input from all child
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@ class SparkPlanInfo(
val metrics: Seq[SQLMetricInfo]) {

override def hashCode(): Int = {
// hashCode of simpleString should be good enough to distinguish the plans from each other
// within a plan
simpleString.hashCode
Copy link
Contributor

Choose a reason for hiding this comment

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

why doesn't this use the same fields as equals? comment

Copy link
Contributor

Choose a reason for hiding this comment

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

Was the intent here to avoid an expensive-to-compute recursive hashcode over children? If so, would memoization help instead?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

We only need to make sure that those object equal with each other should have the same hashCode, but the hashCode does not need to considering all the members.

Using simpleString here should be enough to have good hashCode (be good enough to distinguish each other within a plan)

}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -37,8 +37,9 @@ abstract class Exchange extends UnaryNode {
}

/**
* A wrapper for reused exchange to have different output, which is required to resolve the
* attributes in following plans.
* A wrapper for reused exchange to have different output, because two exchanges which produce
* logically identical output will have distinct sets of output attribute ids, so we need to
* preserve the original ids because they're what downstream operators are expecting.
*/
case class ReusedExchange(override val output: Seq[Attribute], child: Exchange) extends LeafNode {
Copy link
Contributor

Choose a reason for hiding this comment

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

does this need output or should that just be child.output

Copy link
Contributor Author

Choose a reason for hiding this comment

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

The aggregate will have different output, even they have same result, because the Aggregate will create new ExprID for the output.

Copy link
Contributor

Choose a reason for hiding this comment

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

Nvm.

Copy link
Contributor

Choose a reason for hiding this comment

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

Should ReusedExchange extend Exchange?

Copy link
Contributor

Choose a reason for hiding this comment

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

Yeah, so just to summarize: two exchanges which produce logically identical output will have distinct sets of output attribute ids, so we need to preserve the original ids because they're what downstream operators are expecting.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

No, because Exchange is unary node, ReusedExchange is leaf node.

ReusedExchange is similar to InMemoryColumnarTableScan


Expand Down Expand Up @@ -73,15 +74,15 @@ private[sql] case class ReuseExchange(sqlContext: SQLContext) extends Rule[Spark
val exchanges = mutable.HashMap[StructType, ArrayBuffer[Exchange]]()
Copy link
Contributor

Choose a reason for hiding this comment

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

Won't StructType's equals() and hashCode() methods be affected by field names? What if the two exchanges produce logically equivalent output but assign different names to the output columns? In this case, would that lead to false-negatives when searching for exchanges that have the sameResult?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Good question, it could be false negative.

But usually if two plan have the same result, they should have the same inputs also the same plan and expressions, they should generate the same name (does not include the random ExprId).

Copy link
Contributor

Choose a reason for hiding this comment

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

I suppose we can always follow up on this later if it turns out to be a problem in practice.

plan.transformUp {
case exchange: Exchange =>
// the exchanges that have same results usually also have same schemas (same column names).
val sameSchema = exchanges.getOrElseUpdate(exchange.schema, ArrayBuffer[Exchange]())
val samePlan = sameSchema.find { e =>
exchange.sameResult(e)
}
if (samePlan.isDefined) {
// Keep the output of this exchange, the following plans require that to resolve
// attributes.
val reused = ReusedExchange(exchange.output, samePlan.get)
reused
ReusedExchange(exchange.output, samePlan.get)
} else {
sameSchema += exchange
exchange
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -104,12 +104,12 @@ case class ShuffleExchange(
/**
* Caches the created ShuffleRowRDD so we can reuse that.
*/
private var shuffleRDD: ShuffledRowRDD = null
private var cachedShuffleRDD: ShuffledRowRDD = null

protected override def doExecute(): RDD[InternalRow] = attachTree(this, "execute") {
Copy link
Contributor

Choose a reason for hiding this comment

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

Naive question, but do we need to cache the result after the attachTree?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

attachTree is only used to generate better error message (show the plan), I think it do not matter here.

// Returns the same ShuffleRowRDD if this plan is used by multiple plans.
if (shuffleRDD == null) {
shuffleRDD = coordinator match {
if (cachedShuffleRDD == null) {
cachedShuffleRDD = coordinator match {
case Some(exchangeCoordinator) =>
val shuffleRDD = exchangeCoordinator.postShuffleRDD(this)
Copy link
Contributor

Choose a reason for hiding this comment

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

AFAIK IntelliJ might give a "suspicious variable shadowing" warning RE: this name, since shuffleRDD is also defined as a field on this class.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

will rename it to 'cachedShuffleRDD'

assert(shuffleRDD.partitions.length == newPartitioning.numPartitions)
Expand All @@ -119,7 +119,7 @@ case class ShuffleExchange(
preparePostShuffleRDD(shuffleDependency)
}
}
shuffleRDD
cachedShuffleRDD
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -104,8 +104,7 @@ private[sql] object SparkPlanGraph {
} else {
subgraph.nodes += node
}
// ShuffleExchange or BroadcastExchange
if (name.endsWith("Exchange")) {
if (name == "ShuffleExchange" || name == "BroadcastExchange") {
Copy link
Contributor

Choose a reason for hiding this comment

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

@nongli, just to make sure that your original comment here was addressed: were you worried about this pattern being incomplete if we add a new type of exchange? If that's the case, then the move from name.endsWith("Exchange") to this might make things worse.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I'm glad to revert this change.

Btw, it's hard to protect future change anyway (because of unknown).

exchanges += planInfo -> node
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1341,6 +1341,7 @@ class DataFrameSuite extends QueryTest with SharedSQLContext {
val df = sqlContext.range(100)
val agg1 = df.groupBy().count()
val agg2 = df.groupBy().count()
// two aggregates with different ExprId within them should have same result
agg1.queryExecution.executedPlan.sameResult(agg2.queryExecution.executedPlan)
}

Expand Down