Skip to content

[SPARK-24556][SQL] Always rewrite output partitioning in ReusedExchangeExec and InMemoryTableScanExec #21564

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
yucai wants to merge 8 commits into from
Closed

[SPARK-24556][SQL] Always rewrite output partitioning in ReusedExchangeExec and InMemoryTableScanExec #21564

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
f37139b
[SPARK-24556][SQL] ReusedExchange should rewrite output partitioning …
Jun 14, 2018
0ef99cc
improve
Jun 14, 2018
405ba94
improve import
Jun 14, 2018
9da85e0
rework unit tests
Jun 16, 2018
85dc1bc
upate tests
Jun 16, 2018
7d1c2f2
update tests
Jun 16, 2018
6744a9e
no need union in cache table
Jun 19, 2018
dcd0ce9
update tests
Jun 19, 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
4 changes: 3 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.QueryPlan
import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning}
import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning, RangePartitioning}
import org.apache.spark.sql.execution.{ColumnarBatchScan, LeafExecNode, SparkPlan, WholeStageCodegenExec}
import org.apache.spark.sql.execution.vectorized._
import org.apache.spark.sql.types._
Expand Down Expand Up @@ -170,6 +170,8 @@ case class InMemoryTableScanExec(
override def outputPartitioning: Partitioning = {
relation.cachedPlan.outputPartitioning match {
case h: HashPartitioning => updateAttribute(h).asInstanceOf[HashPartitioning]
case r: RangePartitioning =>
r.copy(ordering = r.ordering.map(updateAttribute(_).asInstanceOf[SortOrder]))

@viirya viirya Jun 14, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Not sure why RangePartitioning isn't included at first.

@mgaido91 mgaido91 Jun 14, 2018 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

why not just updateAttribute(r)?

Moreover, in order to avoid the same issue in the future with other cases, have you considered doing something like:

updateAttribute(relation.cachedPlan.outputPartitioning)

?

@viirya viirya Jun 14, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Not all Partitioning are Expression. Only HashPartitioning and RangePartitioning are.

@yucai yucai Jun 14, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

Good suggestion, thanks @mgaido91.

@viirya Do we need consider below:
PartitioningCollection in InMemoryTableScanExec.outputPartitioning, which is also Expression?
PartitioningCollection and BroadcastPartitioning in ReusedExchangeExec.outputPartitioning?

@mgaido91 mgaido91 Jun 14, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

yes, you're right @viirya , thanks. Then, I'd propose something like:

relation.cachedPlan.outputPartitioning match {
 case e: Expression => updateAttribute(e)
 case other => other
}

what do you think?

@viirya viirya Jun 14, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I think PartitioningCollection is for an operator that has multiple children. BroadcastPartitioning is not Expression.

@viirya viirya Jun 14, 2018 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Hmm, HashPartitioning and RangePartitioning can affect later sorting and shuffle. But for BroadcastPartitioning, seems to me no too much benefit.

@yucai yucai Jun 14, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

PartitioningCollection should be considered. Like below case:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
spark.conf.set("spark.sql.codegen.wholeStage", false)
val df1 = Seq(1 -> "a", 3 -> "c", 2 -> "b").toDF("i", "j").as("t1")
val df2 = Seq(1 -> "a", 3 -> "c", 2 -> "b").toDF("m", "n").as("t2")
val d = df1.join(df2, $"t1.i" === $"t2.m")
d.cache
val d1 = d.as("t3")
val d2 = d.as("t4")
d1.join(d2, $"t3.i" === $"t4.i").explain

SortMergeJoin [i#5], [i#54], Inner
:- InMemoryTableScan [i#5, j#6, m#15, n#16]
:     +- InMemoryRelation [i#5, j#6, m#15, n#16], CachedRDDBuilder
:           +- SortMergeJoin [i#5], [m#15], Inner
:              :- Sort [i#5 ASC NULLS FIRST], false, 0
:              :  +- Exchange hashpartitioning(i#5, 10)
:              :     +- LocalTableScan [i#5, j#6]
:              +- Sort [m#15 ASC NULLS FIRST], false, 0
:                 +- Exchange hashpartitioning(m#15, 10)
:                    +- LocalTableScan [m#15, n#16]
+- Sort [i#54 ASC NULLS FIRST], false, 0
   +- Exchange hashpartitioning(i#54, 10)
      +- InMemoryTableScan [i#54, j#55, m#58, n#59]
            +- InMemoryRelation [i#54, j#55, m#58, n#59], CachedRDDBuilder
                  +- SortMergeJoin [i#5], [m#15], Inner
                     :- Sort [i#5 ASC NULLS FIRST], false, 0
                     :  +- Exchange hashpartitioning(i#5, 10)
                     :     +- LocalTableScan [i#5, j#6]
                     +- Sort [m#15 ASC NULLS FIRST], false, 0
                        +- Exchange hashpartitioning(m#15, 10)
                           +- LocalTableScan [m#15, n#16]

Exchange hashpartitioning(i#54, 10) is extra shuffle.

How do you think?

@viirya viirya Jun 15, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

For PartitioningCollection, I think it is harder to treat it like HashPartitioning and RangePartitioning when replacing attributes.

In above example, PartitioningCollection contains HashPartitioning(i#5) and HashPartitioning(m#15), the output of InMemoryRelation is [i#54, j#55, m#58, n#59]. Can we still replace attributes based on the location of attribute in output?

@yucai yucai Jun 15, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

@viirya From updateAttribute, relation.cachedPlan.output and relation.output one to one.

 private def updateAttribute(expr: Expression): Expression = {
    ....
    val attrMap = AttributeMap(relation.cachedPlan.output.zip(relation.output))
    ....
  }

It means "[i#54, j#55, m#58, n#59]" corresponds to "[i#5, j#6, m#15, n#16]", so we can always replace HashPartitioning(i#5) to HashPartitioning(i#54).
Any idea?

@viirya viirya Jun 15, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Looks correct.

case _ => relation.cachedPlan.outputPartitioning
}
}
Expand Down
4 changes: 3 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/Exchange.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ import org.apache.spark.broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, Expression, SortOrder}
import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning}
import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning, RangePartitioning}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan, UnaryExecNode}
import org.apache.spark.sql.internal.SQLConf
Expand Down Expand Up @@ -71,6 +71,8 @@ case class ReusedExchangeExec(override val output: Seq[Attribute], child: Exchan

override def outputPartitioning: Partitioning = child.outputPartitioning match {
case h: HashPartitioning => h.copy(expressions = h.expressions.map(updateAttr))
case r: RangePartitioning =>
r.copy(ordering = r.ordering.map(updateAttr(_).asInstanceOf[SortOrder]))
case other => other

@viirya viirya Jun 16, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

LGTM

}

Expand Down
11 changes: 11 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2270,4 +2270,15 @@ class DataFrameSuite extends QueryTest with SharedSQLContext {
val mapWithBinaryKey = map(lit(Array[Byte](1.toByte)), lit(1))
checkAnswer(spark.range(1).select(mapWithBinaryKey.getItem(Array[Byte](1.toByte))), Row(1))
}

test("SPARK-24556: ReusedExchange should rewrite output partitioning for RangePartitioning") {

@cloud-fan cloud-fan Jun 15, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

this is not an end-to-end test, let's put it in PlannerSuite and also test cached table.

@cloud-fan cloud-fan Jun 15, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

please also mention cached table in PR title

val df = Seq(1 -> "a").toDF("i", "j")
val df1 = df.as("t1")
val df2 = df.as("t2")
val shuffles = df1.orderBy("j").join(df2.orderBy("j"), $"t1.i" === $"t2.i", "right")
.cache().orderBy($"t2.j").queryExecution.executedPlan.collect {
case e: ShuffleExchangeExec => e
}
assert(shuffles.isEmpty)
}
}