[SPARK-19931][SQL] InMemoryTableScanExec should rewrite output partitioning and ordering when aliasing output attributes #17175

viirya · 2017-03-06T09:11:29Z

What changes were proposed in this pull request?

Now InMemoryTableScanExec simply takes the outputPartitioning and outputOrdering from the associated InMemoryRelation's child.outputPartitioning and outputOrdering.

However, InMemoryTableScanExec can alias the output attributes. In this case, its outputPartitioning and outputOrdering are not correct and its parent operators can't correctly determine its data distribution.

How was this patch tested?

Jenkins tests.

Please review http://spark.apache.org/contributing.html before opening a pull request.

…to ensure no unnecessary shuffle/sort in Datasets.

SparkQA · 2017-03-06T11:12:50Z

Test build #73989 has finished for PR 17175 at commit 6e4eba6.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2017-03-06T11:20:56Z

cc @kiszk @hvanhovell @cloud-fan

cloud-fan · 2017-03-09T07:40:06Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Canonicalize.scala

@@ -43,6 +43,12 @@ object Canonicalize extends {
    case _ => e
  }

+  /** Remove some unnecessary parameters. */
+  private[expressions] def ignoreParameters(e: Expression): Expression = e match {


we can just put this logic in ignoreNamesTypes and rename it.

ok. will update.

cloud-fan · 2017-03-09T07:46:30Z

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala

@@ -78,9 +78,42 @@ case class ProjectExec(projectList: Seq[NamedExpression], child: SparkPlan)
    }
  }

-  override def outputOrdering: Seq[SortOrder] = child.outputOrdering


Sorry I don't get it. Why it was wrong?

Take the plan in PR description as an example. In the Project,

Project [named_struct(_1, _1#83, _2, _2#84) AS _1#105]

Its output is _1#105. Its outputPartitioning is _1#83. But the parent operator of the Project requires partitioning on _1#105._1. Since _1#83.semanticEquals(_1#105._1) is false, additional ShuffleExchange will be added.

outputOrdering is the same issue.

Well, if we wanna support this, we should support all the case. e.g.

Project [(a#1 + b#2) as c#3, b#2]

and then the parent operator requires partitioning on c#3 - b#2. Since (c#3 - b#2).samanticEquals(a#1) is false, additional ShuffleExchange will be added.

This is not trivial and we should spend more time on a holistic solution.

My original thought is to support the cases where requiresChildDistribution refers nested fields created and aliased in Project, as the example I showed.

The example you give is another kind of case. In order to support, if we wannt, at least we need to improve expression canonicalization and partitioning/distribution matching.

nested field is nothing special, just CreateStruct and GetStructField expressions. What we should handle is symmetric expressions, e.g. CreateStruct and GetStructField, Add and Subtract, etc.

It doesn't make sense to only handle CreateStruct and GetStructField specially, and leave others behind.

Yeah, I see. In order to support general cases, I am going first improve expression canonicalization. So we can check semantic equal between (a + b) - b and a.

#17242 is a WIP pr for that.

cloud-fan · 2017-03-10T18:53:58Z

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala

@@ -42,10 +42,34 @@ case class InMemoryTableScanExec(
  override def output: Seq[Attribute] = attributes

  // The cached version does not change the outputPartitioning of the original SparkPlan.
-  override def outputPartitioning: Partitioning = relation.child.outputPartitioning
+  // But the cached version could alias output, so we need to replace output.
+  override def outputPartitioning: Partitioning = {


I think this is valid, shall we only keep this and merge this PR first?

Sorry, do you mean only keeping change of outputPartitioning without outputOrdering?

Oh. Do you mean only keeping the change of InMemoryTableScanExec?

I suppose you mean to keep the change of InMemoryTableScanExec. Other changes are removed. Can you take a look? Thanks.

cloud-fan · 2017-03-13T04:30:53Z

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala

+      relation.child.output.zip(output)
+    )
+    relation.child.outputPartitioning match {
+      case HashPartitioning(expressions, numPartitions) =>


HashPartitioning is an expression, we can simplify it to

case h: HashPartitioning => h.transformExpression { ... }.asInstanceOf[HashPartitioning]

OK. Will update.

cloud-fan · 2017-03-13T04:31:38Z

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala

+      relation.child.output.zip(output)
+    )
+    relation.child.outputOrdering.map { sortOrder =>
+      val newSortExpr = sortOrder.child.transform {


same here, SortOrder is an expression

we can even have a method def updateAttribute(expr: Expression): Expression to do this.

SparkQA · 2017-03-13T05:33:09Z

Test build #74419 has finished for PR 17175 at commit c83919e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-03-13T06:17:34Z

Test build #74424 has started for PR 17175 at commit 454515d.

viirya · 2017-03-13T07:53:13Z

retest this please.

SparkQA · 2017-03-13T09:54:05Z

Test build #74438 has finished for PR 17175 at commit 454515d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-03-14T04:20:45Z

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala

@@ -41,11 +41,31 @@ case class InMemoryTableScanExec(

  override def output: Seq[Attribute] = attributes

+  private def updateAttribute(expr: Expression, attrMap: AttributeMap[Attribute]): Expression =


we can create the attrMap in this method

then when processing outputOrdering, we will create attrMap many times.

outputOrdering should not be very long. This may be ok. I will update it.

SparkQA · 2017-03-14T07:50:01Z

Test build #74496 has finished for PR 17175 at commit 1fc023c.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2017-03-14T10:57:29Z

retest this please.

cloud-fan · 2017-03-14T11:28:02Z

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala

+      relation.child.output.zip(output)
+    )
+    expr.transform {
+      case attr: Attribute if attrMap.contains(attr) => attrMap.get(attr).get


nit: case attr: Attribute => attrMap.getOrElse(attr, attr)

SparkQA · 2017-03-14T13:08:59Z

Test build #74511 has finished for PR 17175 at commit 1fc023c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-03-14T15:11:25Z

Test build #74518 has finished for PR 17175 at commit ef918da.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-03-15T06:32:55Z

...core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala

+      .repartition(col("_1")).sortWithinPartitions(col("_1")).persist
+    val ds2 = Seq((0, 0), (1, 1)).toDS
+      .repartition(col("_1")).sortWithinPartitions(col("_1")).persist
+    val joined = ds1.joinWith(ds2, ds1("_1") === ds2("_1"))


somehow my comment is lost, let me type it again: why we need to test join here? The test logic below seems have nothing to do with join.

The join is there to force one underlying relation to alias the output.

I think this only happens for self-join?

Because the two datasets cached have the same logical plan, it is a self-join actually.

can we use DataFrame? The code here only use DataFrame features. We should also add some comments to explain why we join here.

ok. I updated it.

cloud-fan · 2017-03-15T06:33:08Z

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala

@@ -41,11 +41,28 @@ case class InMemoryTableScanExec(

  override def output: Seq[Attribute] = attributes

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


can these in one line?

…sary-shuffle

cloud-fan · 2017-03-15T08:52:16Z

LGTM, pending test

SparkQA · 2017-03-15T09:05:00Z

Test build #74587 has finished for PR 17175 at commit b4d5d0f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-03-15T09:34:34Z

Test build #74591 has finished for PR 17175 at commit b25156f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-03-16T00:19:19Z

thank, merging to master

Rewrite physical Project operator's output partitioning and ordering …

6e4eba6

…to ensure no unnecessary shuffle/sort in Datasets.

cloud-fan reviewed Mar 9, 2017

View reviewed changes

cloud-fan reviewed Mar 10, 2017

View reviewed changes

Keep the change of InMemoryTableScanExec.

c83919e

viirya changed the title ~~[SPARK-19468][SQL] InMemoryTableScanExec should rewrite output partitioning and ordering when aliasing output attributes~~ [SPARK-19931][SQL] InMemoryTableScanExec should rewrite output partitioning and ordering when aliasing output attributes Mar 13, 2017

cloud-fan reviewed Mar 13, 2017

View reviewed changes

Refactor the change.

454515d

cloud-fan reviewed Mar 14, 2017

View reviewed changes

Move attrMap into updateAttribute.

1fc023c

cloud-fan reviewed Mar 14, 2017

View reviewed changes

Address comment.

ef918da

cloud-fan reviewed Mar 15, 2017

View reviewed changes

viirya added 2 commits March 15, 2017 06:54

Merge remote-tracking branch 'upstream/master' into ensure-no-unneces…

d8e8ae0

…sary-shuffle

For comment.

b4d5d0f

Use DataFrame and add comment to test.

b25156f

asfgit closed this in 7d734a6 Mar 16, 2017

viirya deleted the ensure-no-unnecessary-shuffle branch December 27, 2023 18:34

		@@ -41,11 +41,31 @@ case class InMemoryTableScanExec(

		override def output: Seq[Attribute] = attributes

		private def updateAttribute(expr: Expression, attrMap: AttributeMap[Attribute]): Expression =

[SPARK-19931][SQL] InMemoryTableScanExec should rewrite output partitioning and ordering when aliasing output attributes #17175

[SPARK-19931][SQL] InMemoryTableScanExec should rewrite output partitioning and ordering when aliasing output attributes #17175

Conversation

viirya commented Mar 6, 2017 • edited

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Mar 6, 2017

viirya commented Mar 6, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan Mar 9, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan Mar 13, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Mar 13, 2017

SparkQA commented Mar 13, 2017

viirya commented Mar 13, 2017

SparkQA commented Mar 13, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Mar 14, 2017

viirya commented Mar 14, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Mar 14, 2017

SparkQA commented Mar 14, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Mar 15, 2017

SparkQA commented Mar 15, 2017

SparkQA commented Mar 15, 2017

cloud-fan commented Mar 16, 2017

viirya commented Mar 6, 2017 •

edited

cloud-fan Mar 9, 2017 •

edited

cloud-fan Mar 13, 2017 •

edited