[SPARK-30298][SQL] Respect aliases in output partitioning of projects and aggregates #26943

imback82 · 2019-12-19T06:06:39Z

What changes were proposed in this pull request?

Currently, in the following scenario, bucket join is not utilized:

val df = (0 until 20).map(i => (i, i)).toDF("i", "j").as("df")
df.write.format("parquet").bucketBy(8, "i").saveAsTable("t")
sql("CREATE VIEW v AS SELECT * FROM t")
sql("SELECT * FROM t a JOIN v b ON a.i = b.i").explain

== Physical Plan ==
*(4) SortMergeJoin [i#13], [i#15], Inner
:- *(1) Sort [i#13 ASC NULLS FIRST], false, 0
:  +- *(1) Project [i#13, j#14]
:     +- *(1) Filter isnotnull(i#13)
:        +- *(1) ColumnarToRow
:           +- FileScan parquet default.t[i#13,j#14] Batched: true, DataFilters: [isnotnull(i#13)], Format: Parquet, Location: InMemoryFileIndex[file:..., PartitionFilters: [], PushedFilters: [IsNotNull(i)], ReadSchema: struct<i:int,j:int>, SelectedBucketsCount: 8 out of 8
+- *(3) Sort [i#15 ASC NULLS FIRST], false, 0
   +- Exchange hashpartitioning(i#15, 8), true, [id=#64] <----- Exchange node introduced
      +- *(2) Project [i#13 AS i#15, j#14 AS j#16]
         +- *(2) Filter isnotnull(i#13)
            +- *(2) ColumnarToRow
               +- FileScan parquet default.t[i#13,j#14] Batched: true, DataFilters: [isnotnull(i#13)], Format: Parquet, Location: InMemoryFileIndex[file:..., PartitionFilters: [], PushedFilters: [IsNotNull(i)], ReadSchema: struct<i:int,j:int>, SelectedBucketsCount: 8 out of 8

Notice that Exchange is present. This is because Project introduces aliases and outputPartitioning and requiredChildDistribution do not consider aliases while considering bucket join in EnsureRequirements. This PR addresses to allow this scenario.

Why are the changes needed?

This allows bucket join to be utilized in the above example.

Does this PR introduce any user-facing change?

Yes, now with the fix, the explain out is as follows:

== Physical Plan ==
*(3) SortMergeJoin [i#13], [i#15], Inner
:- *(1) Sort [i#13 ASC NULLS FIRST], false, 0
:  +- *(1) Project [i#13, j#14]
:     +- *(1) Filter isnotnull(i#13)
:        +- *(1) ColumnarToRow
:           +- FileScan parquet default.t[i#13,j#14] Batched: true, DataFilters: [isnotnull(i#13)], Format: Parquet, Location: InMemoryFileIndex[file:.., PartitionFilters: [], PushedFilters: [IsNotNull(i)], ReadSchema: struct<i:int,j:int>, SelectedBucketsCount: 8 out of 8
+- *(2) Sort [i#15 ASC NULLS FIRST], false, 0
   +- *(2) Project [i#13 AS i#15, j#14 AS j#16]
      +- *(2) Filter isnotnull(i#13)
         +- *(2) ColumnarToRow
            +- FileScan parquet default.t[i#13,j#14] Batched: true, DataFilters: [isnotnull(i#13)], Format: Parquet, Location: InMemoryFileIndex[file:.., PartitionFilters: [], PushedFilters: [IsNotNull(i)], ReadSchema: struct<i:int,j:int>, SelectedBucketsCount: 8 out of 8

Note that the Exchange is no longer present.

How was this patch tested?

SparkQA · 2019-12-19T08:05:02Z

Test build #115540 has finished for PR 26943 at commit d3559b4.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

imback82 · 2019-12-19T08:12:16Z

retest this please

SparkQA · 2019-12-19T11:53:08Z

Test build #115545 has finished for PR 26943 at commit d3559b4.

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

imback82 · 2019-12-19T15:05:34Z

retest this please

SparkQA · 2019-12-19T19:25:20Z

Test build #115576 has finished for PR 26943 at commit d3559b4.

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

maropu · 2019-12-20T00:17:35Z

Have you checked SPARK-25951, SPARK-19981, and SPARK-19468? I think the other project-like operators (e.g., aggregate) has the same issue. see also #22957

sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala

imback82 · 2019-12-20T04:53:16Z

Thanks @maropu for the info! Adding the same outputPartitioning logic to HashAggregateExec correctly handles cases reported in #22957 and #17400. I can make this more generic similar to what was proposed in those PRs. But could you explain why those two PRs never got merged? Especially, this PR seems to be very similar to #17400 in terms of how Alias is being removed. :)

maropu · 2019-12-20T05:28:33Z

I think we just couldn't reach a consensus about how-to-fix for this issue at that time.

imback82 · 2019-12-20T17:18:44Z

I see. @maropu Do you mind if I take your ideas from #17400 like AliasAwareOutputPartitioning and put them here? I would like to try one more time with the fix if possible.

maropu · 2019-12-20T23:26:17Z

yea, of course not! You can feel free to take them over.

imback82 · 2020-01-03T01:02:37Z

@cloud-fan @gatorsmile @viirya This addresses the same issues brought up in #22957 and #17400. I understand those two PRs didn't get merged, but wanted to give another shot at it. (We had few customers asking why bucket join was not respected when aliases were used). Could you help reviewing? Thanks in advance!

SparkQA · 2020-01-03T03:44:12Z

Test build #116062 has finished for PR 26943 at commit c24789d.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait AliasAwareOutputPartitioning extends UnaryExecNode

imback82 · 2020-01-03T03:48:28Z

retest this please

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

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

SparkQA · 2020-01-03T07:45:45Z

Test build #116066 has finished for PR 26943 at commit c24789d.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait AliasAwareOutputPartitioning extends UnaryExecNode

cloud-fan · 2020-01-03T08:03:30Z

looks fine to me. Shall we also consider some corner cases like a + 1 as b and the child's output partitioning is hash(a + 1)?

SparkQA · 2020-01-07T05:19:43Z

Test build #116197 has finished for PR 26943 at commit fcd2186.

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

SparkQA · 2020-01-07T05:30:01Z

Test build #116198 has finished for PR 26943 at commit 1762a96.

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

maropu · 2020-01-07T23:54:42Z

I'll check later and thanks for the rework, @imback82! cc: @mgaido91

cloud-fan · 2020-01-08T09:15:24Z

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

+
+  final override def outputPartitioning: Partitioning = {
+    child.outputPartitioning match {
+      case HashPartitioning(expressions, numPartitions) =>


a corner case is that: the child output partitioning is a + 1 and the project list has a + 1 as b, then the final output partitioning should be b.

I'm not sure how common it is, maybe it's fine to ignore it.

maropu · 2020-01-08T11:30:44Z

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

+          case other => other
+        }
+        HashPartitioning(newExpressions, numPartitions)
+      case other => other


How about the other partitioning cases, e.g., range?

PartitioningCollection is constructed as PartitioningCollection(Seq(left.outputPartitioning, right.outputPartitioning)), so aliases should have been already removed if the partitioning was HashPartitioning. But we could add one similar to your solution (https://github.com/apache/spark/pull/17400/files#diff-342789ab9c8c0154b412dd1c719c9397R82-R86) for future proof.

For RangePartitioning, your change (https://github.com/apache/spark/pull/17400/files#diff-342789ab9c8c0154b412dd1c719c9397R72-R78) makes sense, but I couldn't come up with an actual example to test against. Do you have one in mind?

maropu · 2020-01-08T11:42:22Z

sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala

+    withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "0") {
+      withTable("t") {
+        withView("v") {
+          val df = (0 until 20).map(i => (i, i)).toDF("i", "j").as("df")


nit: We need .as("df") for this test? I think you can just write it like;

spark.range(20).selectExpr("id as i", "id as j").write.bucketBy(8, "I").saveAsTable("t")

Thanks, updated.

maropu · 2020-01-08T11:44:24Z

sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala

+          val plan1 = sql("SELECT * FROM t a JOIN t b ON a.i = b.i").queryExecution.executedPlan
+          assert(plan1.collect { case exchange: ShuffleExchangeExec => exchange }.isEmpty)
+
+          val plan2 = sql("SELECT * FROM t a JOIN v b ON a.i = b.i").queryExecution.executedPlan


What does this test means? This test can improve the test coverage?

I will remove plan1 which is a benign case.

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

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

SparkQA · 2020-01-23T08:05:02Z

Test build #117276 has finished for PR 26943 at commit b877de7.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-01-23T08:17:37Z

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

+            |  (SELECT key AS k from df2) t2
+            |ON t1.k = t2.k
+          """.stripMargin).queryExecution.executedPlan
+        val exchanges = planned.collect { case s: ShuffleExchangeExec => s }


I was confused about why only one shuffle, then realized it's exchange reuse.

Can we join different data frames? e.g. spark.range(10) and spark.range(20).

Thanks for pointing that out. I updated it and it now generates two ShuffleExchangeExec instead of four.

cloud-fan · 2020-01-23T08:17:56Z

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

+            |  (SELECT key + 1 AS k2 from df2) t2
+            |ON t1.k1 = t2.k2
+            |""".stripMargin).queryExecution.executedPlan
+        val exchanges = planned.collect { case s: ShuffleExchangeExec => s }


cloud-fan

LGTM except a minor comments for the test

SparkQA · 2020-01-23T22:54:40Z

Test build #117319 has finished for PR 26943 at commit fbafedf.

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

maropu

Looks nice and thanks for the work, @imback82 !

maropu · 2020-01-23T23:23:30Z

Thanks! Merged to master.

maropu · 2020-01-23T23:27:02Z

@imback82 I think outputOrdering has the same issue and do you wanna work on that, too?
#17400 (comment)

imback82 · 2020-01-24T00:44:42Z

@imback82 I think outputOrdering has the same issue and do you wanna work on that, too?
#17400 (comment)

Yes, I will work on it.

Thanks @cloud-fan and @maropu for review and guidance!

maropu · 2020-01-24T01:35:10Z

Yes, I will work on it.

Thanks in advance, @imback82 !

maropu · 2020-03-06T00:41:54Z

@imback82 I think outputOrdering has the same issue and do you wanna work on that, too?
#17400 (comment)

Yes, I will work on it.

@imback82 You're still working on that?

imback82 · 2020-03-06T00:55:57Z

@imback82 You're still working on that?

Yes! I tried the failing example in the comment, but that was working fine in the latest Spark. I will look into this further.

maropu · 2020-03-06T02:28:04Z

yea, thanks!

imback82 · 2020-03-07T05:28:17Z

@maropu I created a PR for outputOrdering: #27842

imback82 added 4 commits November 15, 2019 19:08

initial checkin

7711ced

Merge branch 'master' into bucket_alias

31191b8

Merge branch 'master' into bucket_alias

fa53ccf

clean up

d3559b4

dongjoon-hyun added the SQL label Dec 19, 2019

maropu reviewed Dec 20, 2019

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala Outdated Show resolved Hide resolved

imback82 added 2 commits January 2, 2020 14:19

Merge branch 'master' into bucket_alias

2286b9e

address comments

c24789d

cloud-fan reviewed Jan 3, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputPartitioning.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Jan 3, 2020

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala Outdated Show resolved Hide resolved

imback82 added 3 commits January 6, 2020 15:49

Merge remote-tracking branch 'upstream/master' into bucket_alias

5d0244f

address PR comments

fcd2186

fix tests

1762a96

cloud-fan reviewed Jan 8, 2020

View reviewed changes

maropu reviewed Jan 8, 2020

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala Show resolved Hide resolved

maropu reviewed Jan 8, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputPartitioning.scala Show resolved Hide resolved

imback82 added 3 commits January 15, 2020 10:33

Merge branch 'master' into bucket_alias

bc2d072

Merge branch 'master' into bucket_alias

323d4a7

address PR comments

b877de7

cloud-fan reviewed Jan 23, 2020

View reviewed changes

cloud-fan approved these changes Jan 23, 2020

View reviewed changes

address PR comments

fbafedf

maropu changed the title ~~[SPARK-30298][SQL] Bucket join should work for self-join with views~~ [SPARK-30298][SQL] Respect aliases in output partitioning of projects and aggregates Jan 23, 2020

maropu approved these changes Jan 23, 2020

View reviewed changes

maropu closed this in 4847f73 Jan 23, 2020

imback82 mentioned this pull request Jul 17, 2020

Add Alias column support in join condition for JoinIndex rule microsoft/hyperspace#25

Closed

[SPARK-30298][SQL] Respect aliases in output partitioning of projects and aggregates #26943

[SPARK-30298][SQL] Respect aliases in output partitioning of projects and aggregates #26943

Conversation

imback82 commented Dec 19, 2019

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

SparkQA commented Dec 19, 2019

imback82 commented Dec 19, 2019

SparkQA commented Dec 19, 2019

imback82 commented Dec 19, 2019

SparkQA commented Dec 19, 2019

maropu commented Dec 20, 2019 • edited Loading

imback82 commented Dec 20, 2019 • edited Loading

maropu commented Dec 20, 2019 • edited Loading

imback82 commented Dec 20, 2019

maropu commented Dec 20, 2019 • edited Loading

imback82 commented Jan 3, 2020

SparkQA commented Jan 3, 2020

imback82 commented Jan 3, 2020

SparkQA commented Jan 3, 2020

cloud-fan commented Jan 3, 2020

SparkQA commented Jan 7, 2020

SparkQA commented Jan 7, 2020

maropu commented Jan 7, 2020

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

SparkQA commented Jan 23, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan left a comment

Choose a reason for hiding this comment

SparkQA commented Jan 23, 2020

maropu left a comment

Choose a reason for hiding this comment

maropu commented Jan 23, 2020

maropu commented Jan 23, 2020

imback82 commented Jan 24, 2020 • edited Loading

maropu commented Jan 24, 2020

maropu commented Mar 6, 2020 • edited Loading

imback82 commented Mar 6, 2020 • edited Loading

maropu commented Mar 6, 2020

imback82 commented Mar 7, 2020

maropu commented Dec 20, 2019 •

edited

Loading

imback82 commented Dec 20, 2019 •

edited

Loading

maropu commented Dec 20, 2019 •

edited

Loading

maropu commented Dec 20, 2019 •

edited

Loading

imback82 commented Jan 24, 2020 •

edited

Loading

maropu commented Mar 6, 2020 •

edited

Loading

imback82 commented Mar 6, 2020 •

edited

Loading