You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "viirya (via GitHub)" <gi...@apache.org> on 2025/07/22 21:34:44 UTC
[PR] [SPARK-XXXXX][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
viirya opened a new pull request, #51623:
URL: https://github.com/apache/spark/pull/51623
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya closed pull request #51623: [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators
URL: https://github.com/apache/spark/pull/51623
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2233253746
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,8 +701,42 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
Review Comment:
IMO that's a serious drawback. But if we can fix it in a follow-up PR right after this PR then I'm ok with merging. Or just update this PR with you local changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3121399568
> Thanks, @viirya Can you please point me which tests cover the remaining three scenarios?
>
> ```
> // Core Functionality
> test("union partitioning - different partitioners") {
> // Covers: Different partitioner scenarios
> }
>
> // Mixed Partitioners
> test("union partitioning - mixed partitioners") {
> // Covers: Mixed partitioner scenarios
> }
>
> // Error Handling
> test("union partitioning error handling") {
> // Covers: Error scenarios and fallback behavior
> }
> ```
`HiveCompatibilitySuite`'s `add_part_multiple` covers union with commands:
https://github.com/viirya/spark-1/actions/runs/16457342432/job/46519111118
`DynamicPartitionPruningHiveScanSuiteAEOn`'s `SPARK-39338: Remove dynamic pruning subquery if pruningKey's references is empty` covers error handling case:
https://github.com/viirya/spark-1/actions/runs/16457342432/job/46519111126
I think most union tests cover mixed partitioner case, because most union queries don't have same partitioner.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2233921319
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,8 +699,80 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
- protected override def doExecute(): RDD[InternalRow] =
- sparkContext.union(children.map(_.execute()))
+ /**
+ * Returns the output partitionings of the children, with the attributes converted to
+ * the first child's attributes at the same position.
+ */
+ private def prepareOutputPartitioning(): Seq[Partitioning] = {
+ val attributesMap = children.map(_.output).transpose.flatMap { attrs =>
Review Comment:
Maybe separate maps per columns would be safer if `output` of some of the `children.tail` contains duplicates and the those duplicates map to different attributes from `children.head.output`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-XXXXX][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2223903760
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,6 +699,18 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
+
+ override def outputPartitioning: Partitioning = {
+ val nonEmptyRdds = childrenRDDs.filter(!_.partitions.isEmpty)
Review Comment:
Got it~ Thank you for the explanation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3109316034
Hmm, there are a few test failures, I will take a look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3121350571
> @viirya Should unit tests be added in DataFrameSetOperationsSuite.scala to cover the following scenarios?
Those cases are covered by existing tests. For example, there were test failures on union with commands before. I added the logic to handle command case to pass these tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2232821528
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,8 +701,42 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
Review Comment:
Actaully, does this mean that children executions are triggered to get `outputPartitioning` of an `Union`?
E.g. a simple explain to show the physical plan now can trigger execuion of union children?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2224562572
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,6 +699,18 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
Review Comment:
Oh, there is a problem. I looked at the test failures.
There are operators like `SubqueryAdaptiveBroadcastExec`, they are not implemented `execute` method. So we cannot get their rdd during planning time.
Although in theory we can just skip those operators, the problem is Spark doesn't provide a way to know if an operator supports `execute` or not.
Currently I checked the pattern bits to detect unsupported operators.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2233260214
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,8 +701,42 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
Review Comment:
Thanks. I can check it tomorrow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-XXXXX][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2223881294
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -1606,11 +1606,15 @@ class SparkContext(config: SparkConf) extends Logging {
new ReliableCheckpointRDD[T](this, path)
}
+ protected[spark] def isPartitionerAwareUnion[T: ClassTag](rdds: Seq[RDD[T]]): Boolean = {
Review Comment:
Could you add a comment about the assumption, `rdds.filter(!_.partitions.isEmpty)`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3128116558
Thanks @dongjoon-hyun @peter-toth @violetnspct
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2233105944
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,8 +701,42 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
Review Comment:
I actually have the next PR ready locally. After this gets merged, I will open a new PR to improve it and get rid of this `execute` calls.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2226554231
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -5993,6 +5993,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val UNION_OUTPUT_PARTITIONING =
+ buildConf("spark.sql.unionOutputPartitioning")
+ .internal()
+ .doc("When set to true, the output partitioning of UnionExec will be the same as the " +
+ "input partitioning if its children have same partitioner. Otherwise, it will be a " +
+ "default partitioning.")
+ .version("4.1.0")
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
For safety, added an internal config for it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2232350704
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,8 +701,42 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
+
+ override def outputPartitioning: Partitioning = {
+ if (conf.getConf(SQLConf.UNION_OUTPUT_PARTITIONING)) {
+ // Commands like `AppendDataExec` have side effects when creating RDDs, so we
+ // cannot call `execute` on them to determine the partitioning.
+ if (children.exists(_.containsPattern(COMMAND))) {
+ return super.outputPartitioning
+ }
+
+ try {
+ val nonEmptyRdds = childrenRDDs.filter(!_.partitions.isEmpty)
+ if (sparkContext.isPartitionerAwareUnion(nonEmptyRdds)) {
Review Comment:
Actually this only covers limited cases like reused shuffles they have the same partitioner. I would like to extend this to SQL cases, i.e., the `outputPartitioning` is same or compatible for Union's children. But I will leave to follow up works.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2234114849
##########
sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q14a/explain.txt:
##########
@@ -1,129 +1,168 @@
== Physical Plan ==
-TakeOrderedAndProject (125)
-+- * HashAggregate (124)
- +- Exchange (123)
- +- * HashAggregate (122)
- +- Union (121)
- :- * HashAggregate (100)
- : +- Exchange (99)
- : +- * HashAggregate (98)
- : +- Union (97)
- : :- * Filter (66)
- : : +- * HashAggregate (65)
- : : +- Exchange (64)
- : : +- * HashAggregate (63)
- : : +- * Project (62)
- : : +- * BroadcastHashJoin Inner BuildRight (61)
- : : :- * Project (59)
- : : : +- * BroadcastHashJoin Inner BuildRight (58)
- : : : :- * BroadcastHashJoin LeftSemi BuildRight (51)
- : : : : :- * Filter (3)
- : : : : : +- * ColumnarToRow (2)
- : : : : : +- Scan parquet spark_catalog.default.store_sales (1)
- : : : : +- BroadcastExchange (50)
- : : : : +- * Project (49)
- : : : : +- * BroadcastHashJoin Inner BuildRight (48)
- : : : : :- * Filter (6)
- : : : : : +- * ColumnarToRow (5)
- : : : : : +- Scan parquet spark_catalog.default.item (4)
- : : : : +- BroadcastExchange (47)
- : : : : +- * BroadcastHashJoin LeftSemi BuildRight (46)
- : : : : :- * HashAggregate (35)
- : : : : : +- Exchange (34)
- : : : : : +- * HashAggregate (33)
- : : : : : +- * Project (32)
- : : : : : +- * BroadcastHashJoin Inner BuildRight (31)
- : : : : : :- * Project (29)
- : : : : : : +- * BroadcastHashJoin Inner BuildRight (28)
- : : : : : : :- * Filter (9)
- : : : : : : : +- * ColumnarToRow (8)
- : : : : : : : +- Scan parquet spark_catalog.default.store_sales (7)
- : : : : : : +- BroadcastExchange (27)
- : : : : : : +- * BroadcastHashJoin LeftSemi BuildRight (26)
- : : : : : : :- * Filter (12)
- : : : : : : : +- * ColumnarToRow (11)
- : : : : : : : +- Scan parquet spark_catalog.default.item (10)
- : : : : : : +- BroadcastExchange (25)
- : : : : : : +- * Project (24)
- : : : : : : +- * BroadcastHashJoin Inner BuildRight (23)
- : : : : : : :- * Project (21)
- : : : : : : : +- * BroadcastHashJoin Inner BuildRight (20)
- : : : : : : : :- * Filter (15)
- : : : : : : : : +- * ColumnarToRow (14)
- : : : : : : : : +- Scan parquet spark_catalog.default.catalog_sales (13)
- : : : : : : : +- BroadcastExchange (19)
- : : : : : : : +- * Filter (18)
- : : : : : : : +- * ColumnarToRow (17)
- : : : : : : : +- Scan parquet spark_catalog.default.item (16)
- : : : : : : +- ReusedExchange (22)
- : : : : : +- ReusedExchange (30)
- : : : : +- BroadcastExchange (45)
- : : : : +- * Project (44)
- : : : : +- * BroadcastHashJoin Inner BuildRight (43)
- : : : : :- * Project (41)
- : : : : : +- * BroadcastHashJoin Inner BuildRight (40)
- : : : : : :- * Filter (38)
- : : : : : : +- * ColumnarToRow (37)
- : : : : : : +- Scan parquet spark_catalog.default.web_sales (36)
- : : : : : +- ReusedExchange (39)
- : : : : +- ReusedExchange (42)
- : : : +- BroadcastExchange (57)
- : : : +- * BroadcastHashJoin LeftSemi BuildRight (56)
- : : : :- * Filter (54)
- : : : : +- * ColumnarToRow (53)
- : : : : +- Scan parquet spark_catalog.default.item (52)
- : : : +- ReusedExchange (55)
- : : +- ReusedExchange (60)
- : :- * Filter (81)
- : : +- * HashAggregate (80)
- : : +- Exchange (79)
- : : +- * HashAggregate (78)
- : : +- * Project (77)
- : : +- * BroadcastHashJoin Inner BuildRight (76)
- : : :- * Project (74)
- : : : +- * BroadcastHashJoin Inner BuildRight (73)
- : : : :- * BroadcastHashJoin LeftSemi BuildRight (71)
- : : : : :- * Filter (69)
- : : : : : +- * ColumnarToRow (68)
- : : : : : +- Scan parquet spark_catalog.default.catalog_sales (67)
- : : : : +- ReusedExchange (70)
- : : : +- ReusedExchange (72)
- : : +- ReusedExchange (75)
- : +- * Filter (96)
- : +- * HashAggregate (95)
- : +- Exchange (94)
- : +- * HashAggregate (93)
- : +- * Project (92)
- : +- * BroadcastHashJoin Inner BuildRight (91)
- : :- * Project (89)
- : : +- * BroadcastHashJoin Inner BuildRight (88)
- : : :- * BroadcastHashJoin LeftSemi BuildRight (86)
- : : : :- * Filter (84)
- : : : : +- * ColumnarToRow (83)
- : : : : +- Scan parquet spark_catalog.default.web_sales (82)
- : : : +- ReusedExchange (85)
- : : +- ReusedExchange (87)
- : +- ReusedExchange (90)
- :- * HashAggregate (105)
- : +- Exchange (104)
- : +- * HashAggregate (103)
- : +- * HashAggregate (102)
- : +- ReusedExchange (101)
- :- * HashAggregate (110)
- : +- Exchange (109)
- : +- * HashAggregate (108)
- : +- * HashAggregate (107)
- : +- ReusedExchange (106)
- :- * HashAggregate (115)
- : +- Exchange (114)
- : +- * HashAggregate (113)
- : +- * HashAggregate (112)
- : +- ReusedExchange (111)
- +- * HashAggregate (120)
- +- Exchange (119)
- +- * HashAggregate (118)
- +- * HashAggregate (117)
- +- ReusedExchange (116)
+TakeOrderedAndProject (164)
++- * HashAggregate (163)
+ +- Exchange (162)
+ +- * HashAggregate (161)
+ +- Union (160)
+ :- * HashAggregate (99)
+ : +- * HashAggregate (98)
+ : +- Union (97)
+ : :- * Filter (66)
+ : : +- * HashAggregate (65)
+ : : +- Exchange (64)
+ : : +- * HashAggregate (63)
+ : : +- * Project (62)
+ : : +- * BroadcastHashJoin Inner BuildRight (61)
+ : : :- * Project (59)
+ : : : +- * BroadcastHashJoin Inner BuildRight (58)
+ : : : :- * BroadcastHashJoin LeftSemi BuildRight (51)
+ : : : : :- * Filter (3)
+ : : : : : +- * ColumnarToRow (2)
+ : : : : : +- Scan parquet spark_catalog.default.store_sales (1)
+ : : : : +- BroadcastExchange (50)
+ : : : : +- * Project (49)
+ : : : : +- * BroadcastHashJoin Inner BuildRight (48)
+ : : : : :- * Filter (6)
+ : : : : : +- * ColumnarToRow (5)
+ : : : : : +- Scan parquet spark_catalog.default.item (4)
+ : : : : +- BroadcastExchange (47)
+ : : : : +- * BroadcastHashJoin LeftSemi BuildRight (46)
+ : : : : :- * HashAggregate (35)
+ : : : : : +- Exchange (34)
+ : : : : : +- * HashAggregate (33)
+ : : : : : +- * Project (32)
+ : : : : : +- * BroadcastHashJoin Inner BuildRight (31)
+ : : : : : :- * Project (29)
+ : : : : : : +- * BroadcastHashJoin Inner BuildRight (28)
+ : : : : : : :- * Filter (9)
+ : : : : : : : +- * ColumnarToRow (8)
+ : : : : : : : +- Scan parquet spark_catalog.default.store_sales (7)
+ : : : : : : +- BroadcastExchange (27)
+ : : : : : : +- * BroadcastHashJoin LeftSemi BuildRight (26)
+ : : : : : : :- * Filter (12)
+ : : : : : : : +- * ColumnarToRow (11)
+ : : : : : : : +- Scan parquet spark_catalog.default.item (10)
+ : : : : : : +- BroadcastExchange (25)
+ : : : : : : +- * Project (24)
+ : : : : : : +- * BroadcastHashJoin Inner BuildRight (23)
+ : : : : : : :- * Project (21)
+ : : : : : : : +- * BroadcastHashJoin Inner BuildRight (20)
+ : : : : : : : :- * Filter (15)
+ : : : : : : : : +- * ColumnarToRow (14)
+ : : : : : : : : +- Scan parquet spark_catalog.default.catalog_sales (13)
+ : : : : : : : +- BroadcastExchange (19)
+ : : : : : : : +- * Filter (18)
+ : : : : : : : +- * ColumnarToRow (17)
+ : : : : : : : +- Scan parquet spark_catalog.default.item (16)
+ : : : : : : +- ReusedExchange (22)
+ : : : : : +- ReusedExchange (30)
+ : : : : +- BroadcastExchange (45)
+ : : : : +- * Project (44)
+ : : : : +- * BroadcastHashJoin Inner BuildRight (43)
+ : : : : :- * Project (41)
+ : : : : : +- * BroadcastHashJoin Inner BuildRight (40)
+ : : : : : :- * Filter (38)
+ : : : : : : +- * ColumnarToRow (37)
+ : : : : : : +- Scan parquet spark_catalog.default.web_sales (36)
+ : : : : : +- ReusedExchange (39)
+ : : : : +- ReusedExchange (42)
+ : : : +- BroadcastExchange (57)
+ : : : +- * BroadcastHashJoin LeftSemi BuildRight (56)
+ : : : :- * Filter (54)
+ : : : : +- * ColumnarToRow (53)
+ : : : : +- Scan parquet spark_catalog.default.item (52)
+ : : : +- ReusedExchange (55)
+ : : +- ReusedExchange (60)
+ : :- * Filter (81)
+ : : +- * HashAggregate (80)
+ : : +- Exchange (79)
+ : : +- * HashAggregate (78)
+ : : +- * Project (77)
+ : : +- * BroadcastHashJoin Inner BuildRight (76)
+ : : :- * Project (74)
+ : : : +- * BroadcastHashJoin Inner BuildRight (73)
+ : : : :- * BroadcastHashJoin LeftSemi BuildRight (71)
+ : : : : :- * Filter (69)
+ : : : : : +- * ColumnarToRow (68)
+ : : : : : +- Scan parquet spark_catalog.default.catalog_sales (67)
+ : : : : +- ReusedExchange (70)
+ : : : +- ReusedExchange (72)
+ : : +- ReusedExchange (75)
+ : +- * Filter (96)
+ : +- * HashAggregate (95)
+ : +- Exchange (94)
+ : +- * HashAggregate (93)
+ : +- * Project (92)
+ : +- * BroadcastHashJoin Inner BuildRight (91)
+ : :- * Project (89)
+ : : +- * BroadcastHashJoin Inner BuildRight (88)
+ : : :- * BroadcastHashJoin LeftSemi BuildRight (86)
+ : : : :- * Filter (84)
+ : : : : +- * ColumnarToRow (83)
+ : : : : +- Scan parquet spark_catalog.default.web_sales (82)
+ : : : +- ReusedExchange (85)
+ : : +- ReusedExchange (87)
+ : +- ReusedExchange (90)
+ :- * HashAggregate (114)
+ : +- Exchange (113)
+ : +- * HashAggregate (112)
+ : +- * HashAggregate (111)
+ : +- * HashAggregate (110)
+ : +- Union (109)
Review Comment:
Hm, interesting. I think it may depend on the cost of the removed `Exchange (99)` and if it can cover the segments:
```
* HashAggregate
+- Union
:- * Filter
: +- * HashAggregate
:- * Filter
: +- * HashAggregate
:- * Filter
: +- * HashAggregate
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-XXXXX][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2223883622
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,6 +699,18 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
+
+ override def outputPartitioning: Partitioning = {
+ val nonEmptyRdds = childrenRDDs.filter(!_.partitions.isEmpty)
Review Comment:
ditto. We can remove this too if `isPartitionerAwareUnion` has the logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2225055629
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -1606,11 +1606,17 @@ class SparkContext(config: SparkConf) extends Logging {
new ReliableCheckpointRDD[T](this, path)
}
+ // Note that input rdds must be all non-empty, i.e., rdds.filter(_.partitions.isEmpty).isEmpty
+ protected[spark] def isPartitionerAwareUnion[T: ClassTag](rdds: Seq[RDD[T]]): Boolean = {
+ assert(!rdds.exists(_.partitions.isEmpty), "Must not have empty RDDs")
+ val partitioners = rdds.flatMap(_.partitioner).toSet
+ rdds.forall(_.partitioner.isDefined) && partitioners.size == 1
Review Comment:
It seems we don't need the `partitioners` set before the forall isDefined check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2232300146
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala:
##########
@@ -145,6 +145,12 @@ case class BroadcastExchangeExec(
Statistics(dataSize, Some(rowCount))
}
+ override def resetMetrics(): Unit = {
+ // no-op
+ // BroadcastExchangeExec after materialized won't be materialized again, so we should not
+ // reset the metrics. Otherwise, we will lose the metrics collected in the broadcast job.
+ }
Review Comment:
I spent a lot time debugging the remaining test failures. When there is broadcast exchange operator, AQE empty relation propagation rule will produce incorrect query plan around it. It is caused by this reset metrics method.
I think it is valuable to be a separate PR: https://github.com/apache/spark/pull/51673
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2231743256
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -5993,6 +5993,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val UNION_OUTPUT_PARTITIONING =
+ buildConf("spark.sql.unionOutputPartitioning")
+ .internal()
+ .doc("When set to true, the output partitioning of UnionExec will be the same as the " +
+ "input partitioning if its children have same partitioner. Otherwise, it will be a " +
+ "default partitioning.")
+ .version("4.1.0")
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
Thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2232606389
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala:
##########
@@ -145,6 +145,12 @@ case class BroadcastExchangeExec(
Statistics(dataSize, Some(rowCount))
}
+ override def resetMetrics(): Unit = {
+ // no-op
+ // BroadcastExchangeExec after materialized won't be materialized again, so we should not
+ // reset the metrics. Otherwise, we will lose the metrics collected in the broadcast job.
+ }
Review Comment:
https://github.com/apache/spark/pull/51673 also has related test case. Let's merge https://github.com/apache/spark/pull/51673 first and I can rebase this later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2233063066
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,8 +701,42 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
Review Comment:
Yes, so this approach has this drawback. So as I mentioned https://github.com/apache/spark/pull/51623#discussion_r2232350704, this doesn't cover SQL cases generally. I plan to extend this to deal with `outputPartitioning` of children, i.e., no need to invoke `execute` on children.
It will be done in follow up works.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2233105944
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,8 +701,42 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
Review Comment:
I actually have the PR ready locally. After this gets merged, I will open a new PR to improve it and get rid of this `execute` calls.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2233257232
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,8 +701,42 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
Review Comment:
Okay, I updated to this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-XXXXX][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2223881294
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -1606,11 +1606,15 @@ class SparkContext(config: SparkConf) extends Logging {
new ReliableCheckpointRDD[T](this, path)
}
+ protected[spark] def isPartitionerAwareUnion[T: ClassTag](rdds: Seq[RDD[T]]): Boolean = {
Review Comment:
Could you add a comment about the assumption, `rdds.filter(!_.partitions.isEmpty)`? Otherwise, it may cause correctness issues later if we use this blindly.
Otherwise, we had better include the assumption inside this method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-XXXXX][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2223905263
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -1606,11 +1606,17 @@ class SparkContext(config: SparkConf) extends Logging {
new ReliableCheckpointRDD[T](this, path)
}
+ // Note that input rdds must be all non-empty, i.e., rdds.filter(_.partitions.isEmpty).isEmpty
+ protected[spark] def isPartitionerAwareUnion[T: ClassTag](rdds: Seq[RDD[T]]): Boolean = {
+ assert(!rdds.exists(_.partitions.isEmpty), "Must not have empty RDDs")
Review Comment:
Nice!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2234186149
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,8 +699,80 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
- protected override def doExecute(): RDD[InternalRow] =
- sparkContext.union(children.map(_.execute()))
+ /**
+ * Returns the output partitionings of the children, with the attributes converted to
+ * the first child's attributes at the same position.
+ */
+ private def prepareOutputPartitioning(): Seq[Partitioning] = {
+ val attributesMap = children.map(_.output).transpose.flatMap { attrs =>
Review Comment:
Makes sense. Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3276993601
I will take a look. Thanks for reporting that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3278810409
@bersprockets I proposed a fix at https://github.com/apache/spark/pull/52305.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "violetnspct (via GitHub)" <gi...@apache.org>.
violetnspct commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3121104468
Should unit tests be added in DataFrameSetOperationsSuite.scala to cover the following scenarios?
```scala
// Core Functionality
test("union partitioning - different partitioners") {
// Covers: Different partitioner scenarios
}
// Mixed Partitioners
test("union partitioning - mixed partitioners") {
// Covers: Mixed partitioner scenarios
}
// Command Handling
test("union partitioning with commands") {
// Covers: Command plan interactions
}
// Error Handling
test("union partitioning error handling") {
// Covers: Error scenarios and fallback behavior
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2234131968
##########
sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q14a/explain.txt:
##########
@@ -1,129 +1,168 @@
== Physical Plan ==
-TakeOrderedAndProject (125)
-+- * HashAggregate (124)
- +- Exchange (123)
- +- * HashAggregate (122)
- +- Union (121)
- :- * HashAggregate (100)
- : +- Exchange (99)
- : +- * HashAggregate (98)
- : +- Union (97)
- : :- * Filter (66)
- : : +- * HashAggregate (65)
- : : +- Exchange (64)
- : : +- * HashAggregate (63)
- : : +- * Project (62)
- : : +- * BroadcastHashJoin Inner BuildRight (61)
- : : :- * Project (59)
- : : : +- * BroadcastHashJoin Inner BuildRight (58)
- : : : :- * BroadcastHashJoin LeftSemi BuildRight (51)
- : : : : :- * Filter (3)
- : : : : : +- * ColumnarToRow (2)
- : : : : : +- Scan parquet spark_catalog.default.store_sales (1)
- : : : : +- BroadcastExchange (50)
- : : : : +- * Project (49)
- : : : : +- * BroadcastHashJoin Inner BuildRight (48)
- : : : : :- * Filter (6)
- : : : : : +- * ColumnarToRow (5)
- : : : : : +- Scan parquet spark_catalog.default.item (4)
- : : : : +- BroadcastExchange (47)
- : : : : +- * BroadcastHashJoin LeftSemi BuildRight (46)
- : : : : :- * HashAggregate (35)
- : : : : : +- Exchange (34)
- : : : : : +- * HashAggregate (33)
- : : : : : +- * Project (32)
- : : : : : +- * BroadcastHashJoin Inner BuildRight (31)
- : : : : : :- * Project (29)
- : : : : : : +- * BroadcastHashJoin Inner BuildRight (28)
- : : : : : : :- * Filter (9)
- : : : : : : : +- * ColumnarToRow (8)
- : : : : : : : +- Scan parquet spark_catalog.default.store_sales (7)
- : : : : : : +- BroadcastExchange (27)
- : : : : : : +- * BroadcastHashJoin LeftSemi BuildRight (26)
- : : : : : : :- * Filter (12)
- : : : : : : : +- * ColumnarToRow (11)
- : : : : : : : +- Scan parquet spark_catalog.default.item (10)
- : : : : : : +- BroadcastExchange (25)
- : : : : : : +- * Project (24)
- : : : : : : +- * BroadcastHashJoin Inner BuildRight (23)
- : : : : : : :- * Project (21)
- : : : : : : : +- * BroadcastHashJoin Inner BuildRight (20)
- : : : : : : : :- * Filter (15)
- : : : : : : : : +- * ColumnarToRow (14)
- : : : : : : : : +- Scan parquet spark_catalog.default.catalog_sales (13)
- : : : : : : : +- BroadcastExchange (19)
- : : : : : : : +- * Filter (18)
- : : : : : : : +- * ColumnarToRow (17)
- : : : : : : : +- Scan parquet spark_catalog.default.item (16)
- : : : : : : +- ReusedExchange (22)
- : : : : : +- ReusedExchange (30)
- : : : : +- BroadcastExchange (45)
- : : : : +- * Project (44)
- : : : : +- * BroadcastHashJoin Inner BuildRight (43)
- : : : : :- * Project (41)
- : : : : : +- * BroadcastHashJoin Inner BuildRight (40)
- : : : : : :- * Filter (38)
- : : : : : : +- * ColumnarToRow (37)
- : : : : : : +- Scan parquet spark_catalog.default.web_sales (36)
- : : : : : +- ReusedExchange (39)
- : : : : +- ReusedExchange (42)
- : : : +- BroadcastExchange (57)
- : : : +- * BroadcastHashJoin LeftSemi BuildRight (56)
- : : : :- * Filter (54)
- : : : : +- * ColumnarToRow (53)
- : : : : +- Scan parquet spark_catalog.default.item (52)
- : : : +- ReusedExchange (55)
- : : +- ReusedExchange (60)
- : :- * Filter (81)
- : : +- * HashAggregate (80)
- : : +- Exchange (79)
- : : +- * HashAggregate (78)
- : : +- * Project (77)
- : : +- * BroadcastHashJoin Inner BuildRight (76)
- : : :- * Project (74)
- : : : +- * BroadcastHashJoin Inner BuildRight (73)
- : : : :- * BroadcastHashJoin LeftSemi BuildRight (71)
- : : : : :- * Filter (69)
- : : : : : +- * ColumnarToRow (68)
- : : : : : +- Scan parquet spark_catalog.default.catalog_sales (67)
- : : : : +- ReusedExchange (70)
- : : : +- ReusedExchange (72)
- : : +- ReusedExchange (75)
- : +- * Filter (96)
- : +- * HashAggregate (95)
- : +- Exchange (94)
- : +- * HashAggregate (93)
- : +- * Project (92)
- : +- * BroadcastHashJoin Inner BuildRight (91)
- : :- * Project (89)
- : : +- * BroadcastHashJoin Inner BuildRight (88)
- : : :- * BroadcastHashJoin LeftSemi BuildRight (86)
- : : : :- * Filter (84)
- : : : : +- * ColumnarToRow (83)
- : : : : +- Scan parquet spark_catalog.default.web_sales (82)
- : : : +- ReusedExchange (85)
- : : +- ReusedExchange (87)
- : +- ReusedExchange (90)
- :- * HashAggregate (105)
- : +- Exchange (104)
- : +- * HashAggregate (103)
- : +- * HashAggregate (102)
- : +- ReusedExchange (101)
- :- * HashAggregate (110)
- : +- Exchange (109)
- : +- * HashAggregate (108)
- : +- * HashAggregate (107)
- : +- ReusedExchange (106)
- :- * HashAggregate (115)
- : +- Exchange (114)
- : +- * HashAggregate (113)
- : +- * HashAggregate (112)
- : +- ReusedExchange (111)
- +- * HashAggregate (120)
- +- Exchange (119)
- +- * HashAggregate (118)
- +- * HashAggregate (117)
- +- ReusedExchange (116)
+TakeOrderedAndProject (164)
++- * HashAggregate (163)
+ +- Exchange (162)
+ +- * HashAggregate (161)
+ +- Union (160)
+ :- * HashAggregate (99)
+ : +- * HashAggregate (98)
+ : +- Union (97)
+ : :- * Filter (66)
+ : : +- * HashAggregate (65)
+ : : +- Exchange (64)
+ : : +- * HashAggregate (63)
+ : : +- * Project (62)
+ : : +- * BroadcastHashJoin Inner BuildRight (61)
+ : : :- * Project (59)
+ : : : +- * BroadcastHashJoin Inner BuildRight (58)
+ : : : :- * BroadcastHashJoin LeftSemi BuildRight (51)
+ : : : : :- * Filter (3)
+ : : : : : +- * ColumnarToRow (2)
+ : : : : : +- Scan parquet spark_catalog.default.store_sales (1)
+ : : : : +- BroadcastExchange (50)
+ : : : : +- * Project (49)
+ : : : : +- * BroadcastHashJoin Inner BuildRight (48)
+ : : : : :- * Filter (6)
+ : : : : : +- * ColumnarToRow (5)
+ : : : : : +- Scan parquet spark_catalog.default.item (4)
+ : : : : +- BroadcastExchange (47)
+ : : : : +- * BroadcastHashJoin LeftSemi BuildRight (46)
+ : : : : :- * HashAggregate (35)
+ : : : : : +- Exchange (34)
+ : : : : : +- * HashAggregate (33)
+ : : : : : +- * Project (32)
+ : : : : : +- * BroadcastHashJoin Inner BuildRight (31)
+ : : : : : :- * Project (29)
+ : : : : : : +- * BroadcastHashJoin Inner BuildRight (28)
+ : : : : : : :- * Filter (9)
+ : : : : : : : +- * ColumnarToRow (8)
+ : : : : : : : +- Scan parquet spark_catalog.default.store_sales (7)
+ : : : : : : +- BroadcastExchange (27)
+ : : : : : : +- * BroadcastHashJoin LeftSemi BuildRight (26)
+ : : : : : : :- * Filter (12)
+ : : : : : : : +- * ColumnarToRow (11)
+ : : : : : : : +- Scan parquet spark_catalog.default.item (10)
+ : : : : : : +- BroadcastExchange (25)
+ : : : : : : +- * Project (24)
+ : : : : : : +- * BroadcastHashJoin Inner BuildRight (23)
+ : : : : : : :- * Project (21)
+ : : : : : : : +- * BroadcastHashJoin Inner BuildRight (20)
+ : : : : : : : :- * Filter (15)
+ : : : : : : : : +- * ColumnarToRow (14)
+ : : : : : : : : +- Scan parquet spark_catalog.default.catalog_sales (13)
+ : : : : : : : +- BroadcastExchange (19)
+ : : : : : : : +- * Filter (18)
+ : : : : : : : +- * ColumnarToRow (17)
+ : : : : : : : +- Scan parquet spark_catalog.default.item (16)
+ : : : : : : +- ReusedExchange (22)
+ : : : : : +- ReusedExchange (30)
+ : : : : +- BroadcastExchange (45)
+ : : : : +- * Project (44)
+ : : : : +- * BroadcastHashJoin Inner BuildRight (43)
+ : : : : :- * Project (41)
+ : : : : : +- * BroadcastHashJoin Inner BuildRight (40)
+ : : : : : :- * Filter (38)
+ : : : : : : +- * ColumnarToRow (37)
+ : : : : : : +- Scan parquet spark_catalog.default.web_sales (36)
+ : : : : : +- ReusedExchange (39)
+ : : : : +- ReusedExchange (42)
+ : : : +- BroadcastExchange (57)
+ : : : +- * BroadcastHashJoin LeftSemi BuildRight (56)
+ : : : :- * Filter (54)
+ : : : : +- * ColumnarToRow (53)
+ : : : : +- Scan parquet spark_catalog.default.item (52)
+ : : : +- ReusedExchange (55)
+ : : +- ReusedExchange (60)
+ : :- * Filter (81)
+ : : +- * HashAggregate (80)
+ : : +- Exchange (79)
+ : : +- * HashAggregate (78)
+ : : +- * Project (77)
+ : : +- * BroadcastHashJoin Inner BuildRight (76)
+ : : :- * Project (74)
+ : : : +- * BroadcastHashJoin Inner BuildRight (73)
+ : : : :- * BroadcastHashJoin LeftSemi BuildRight (71)
+ : : : : :- * Filter (69)
+ : : : : : +- * ColumnarToRow (68)
+ : : : : : +- Scan parquet spark_catalog.default.catalog_sales (67)
+ : : : : +- ReusedExchange (70)
+ : : : +- ReusedExchange (72)
+ : : +- ReusedExchange (75)
+ : +- * Filter (96)
+ : +- * HashAggregate (95)
+ : +- Exchange (94)
+ : +- * HashAggregate (93)
+ : +- * Project (92)
+ : +- * BroadcastHashJoin Inner BuildRight (91)
+ : :- * Project (89)
+ : : +- * BroadcastHashJoin Inner BuildRight (88)
+ : : :- * BroadcastHashJoin LeftSemi BuildRight (86)
+ : : : :- * Filter (84)
+ : : : : +- * ColumnarToRow (83)
+ : : : : +- Scan parquet spark_catalog.default.web_sales (82)
+ : : : +- ReusedExchange (85)
+ : : +- ReusedExchange (87)
+ : +- ReusedExchange (90)
+ :- * HashAggregate (114)
+ : +- Exchange (113)
+ : +- * HashAggregate (112)
+ : +- * HashAggregate (111)
+ : +- * HashAggregate (110)
+ : +- Union (109)
Review Comment:
Yeah. Unfortunately the above happens 4 times in this query.
But still, IMO this PR is a step in the right direction to remove unnecessary exchanges.
In the future some kind of common subplan elimination rule should identify and cache query parts that appear multiple times if computation costs exceed the cost of caching.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "violetnspct (via GitHub)" <gi...@apache.org>.
violetnspct commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3121389365
Thanks, @viirya
Can you please point me which tests cover the remaining three scenarios?
```
// Core Functionality
test("union partitioning - different partitioners") {
// Covers: Different partitioner scenarios
}
// Mixed Partitioners
test("union partitioning - mixed partitioners") {
// Covers: Mixed partitioner scenarios
}
// Error Handling
test("union partitioning error handling") {
// Covers: Error scenarios and fallback behavior
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-XXXXX][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3104911993
cc @peter-toth
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "bersprockets (via GitHub)" <gi...@apache.org>.
bersprockets commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3276650168
@viirya
I'm getting an error that I bisected to this commit (I get the error on this commit, but not on the one before):
```
java.lang.IllegalArgumentException: Can't zip RDDs with unequal numbers of partitions: List(200, 400)
at org.apache.spark.rdd.ZippedPartitionsBaseRDD.getPartitions(ZippedPartitionsRDD.scala:58)
at org.apache.spark.rdd.RDD.$anonfun$partitions$2(RDD.scala:301)
at scala.Option.getOrElse(Option.scala:201)
at org.apache.spark.rdd.RDD.partitions(RDD.scala:297)
at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:49)
```
These are the SQL statements I use to reproduce:
```
# run so that the joins are sortmerge joins
bin/spark-sql --conf spark.sql.autoBroadcastJoinThreshold=-1
create or replace temp view person(id, name, age) as values
(0, "mike", 30),
(1, "jim", 20);
create or replace temp view person_a as
SELECT name, avg(age) as avg_age FROM person GROUP BY name;
create or replace temp view person_b as
SELECT p1.name, p2.avg_age FROM person p1 JOIN person_a p2 ON p1.name = p2.name;
create or replace temp view person_c as
SELECT * FROM person_a UNION SELECT * FROM person_b;
SELECT p1.name, p2.avg_age FROM person_c p1 JOIN person_c p2 ON p1.name = p2.name;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-XXXXX][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2223901261
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -1606,11 +1606,15 @@ class SparkContext(config: SparkConf) extends Logging {
new ReliableCheckpointRDD[T](this, path)
}
+ protected[spark] def isPartitionerAwareUnion[T: ClassTag](rdds: Seq[RDD[T]]): Boolean = {
Review Comment:
Added comment and a check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2224591970
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,6 +700,28 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
+
+ override def outputPartitioning: Partitioning = {
+ try {
+ val nonEmptyRdds = childrenRDDs.filter(!_.partitions.isEmpty)
+ if (sparkContext.isPartitionerAwareUnion(nonEmptyRdds)) {
+ // `isPartitionerAwareUnion` ensures that at least one child is non-empty.
+ children.head.outputPartitioning
+ } else {
+ super.outputPartitioning
+ }
+ } catch {
+ // If any child operator doesn't support `execute`, we cannot determine the
+ // partitioning. Even if it is other exception, we also simply fall back to
+ // the default partitioning. Note that for such cases, it means that these
+ // child operator will be replaced by Spark in query planning later, in other
+ // words, `execute` won't be actually called on them during the execution of
+ // this plan. So we can safely return the default partitioning.
+ case e if NonFatal(e) => super.outputPartitioning
Review Comment:
This handles nodes that don't implement `execute` method. The reason is described like the comment said.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2224562572
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,6 +699,18 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
Review Comment:
Oh, there is a problem. I looked at the test failures.
There are operators like `SubqueryAdaptiveBroadcastExec`, they are not implemented `execute` method. So we cannot get their rdd during planning time.
Although in theory we can just skip those operators, the problem is Spark doesn't provide a way to know if an operator supports `execute` or not.
Currently I checked the pattern bits to detect unsupported operators.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2233869384
##########
sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q14a/explain.txt:
##########
@@ -1,129 +1,168 @@
== Physical Plan ==
-TakeOrderedAndProject (125)
-+- * HashAggregate (124)
- +- Exchange (123)
- +- * HashAggregate (122)
- +- Union (121)
- :- * HashAggregate (100)
- : +- Exchange (99)
- : +- * HashAggregate (98)
- : +- Union (97)
- : :- * Filter (66)
- : : +- * HashAggregate (65)
- : : +- Exchange (64)
- : : +- * HashAggregate (63)
- : : +- * Project (62)
- : : +- * BroadcastHashJoin Inner BuildRight (61)
- : : :- * Project (59)
- : : : +- * BroadcastHashJoin Inner BuildRight (58)
- : : : :- * BroadcastHashJoin LeftSemi BuildRight (51)
- : : : : :- * Filter (3)
- : : : : : +- * ColumnarToRow (2)
- : : : : : +- Scan parquet spark_catalog.default.store_sales (1)
- : : : : +- BroadcastExchange (50)
- : : : : +- * Project (49)
- : : : : +- * BroadcastHashJoin Inner BuildRight (48)
- : : : : :- * Filter (6)
- : : : : : +- * ColumnarToRow (5)
- : : : : : +- Scan parquet spark_catalog.default.item (4)
- : : : : +- BroadcastExchange (47)
- : : : : +- * BroadcastHashJoin LeftSemi BuildRight (46)
- : : : : :- * HashAggregate (35)
- : : : : : +- Exchange (34)
- : : : : : +- * HashAggregate (33)
- : : : : : +- * Project (32)
- : : : : : +- * BroadcastHashJoin Inner BuildRight (31)
- : : : : : :- * Project (29)
- : : : : : : +- * BroadcastHashJoin Inner BuildRight (28)
- : : : : : : :- * Filter (9)
- : : : : : : : +- * ColumnarToRow (8)
- : : : : : : : +- Scan parquet spark_catalog.default.store_sales (7)
- : : : : : : +- BroadcastExchange (27)
- : : : : : : +- * BroadcastHashJoin LeftSemi BuildRight (26)
- : : : : : : :- * Filter (12)
- : : : : : : : +- * ColumnarToRow (11)
- : : : : : : : +- Scan parquet spark_catalog.default.item (10)
- : : : : : : +- BroadcastExchange (25)
- : : : : : : +- * Project (24)
- : : : : : : +- * BroadcastHashJoin Inner BuildRight (23)
- : : : : : : :- * Project (21)
- : : : : : : : +- * BroadcastHashJoin Inner BuildRight (20)
- : : : : : : : :- * Filter (15)
- : : : : : : : : +- * ColumnarToRow (14)
- : : : : : : : : +- Scan parquet spark_catalog.default.catalog_sales (13)
- : : : : : : : +- BroadcastExchange (19)
- : : : : : : : +- * Filter (18)
- : : : : : : : +- * ColumnarToRow (17)
- : : : : : : : +- Scan parquet spark_catalog.default.item (16)
- : : : : : : +- ReusedExchange (22)
- : : : : : +- ReusedExchange (30)
- : : : : +- BroadcastExchange (45)
- : : : : +- * Project (44)
- : : : : +- * BroadcastHashJoin Inner BuildRight (43)
- : : : : :- * Project (41)
- : : : : : +- * BroadcastHashJoin Inner BuildRight (40)
- : : : : : :- * Filter (38)
- : : : : : : +- * ColumnarToRow (37)
- : : : : : : +- Scan parquet spark_catalog.default.web_sales (36)
- : : : : : +- ReusedExchange (39)
- : : : : +- ReusedExchange (42)
- : : : +- BroadcastExchange (57)
- : : : +- * BroadcastHashJoin LeftSemi BuildRight (56)
- : : : :- * Filter (54)
- : : : : +- * ColumnarToRow (53)
- : : : : +- Scan parquet spark_catalog.default.item (52)
- : : : +- ReusedExchange (55)
- : : +- ReusedExchange (60)
- : :- * Filter (81)
- : : +- * HashAggregate (80)
- : : +- Exchange (79)
- : : +- * HashAggregate (78)
- : : +- * Project (77)
- : : +- * BroadcastHashJoin Inner BuildRight (76)
- : : :- * Project (74)
- : : : +- * BroadcastHashJoin Inner BuildRight (73)
- : : : :- * BroadcastHashJoin LeftSemi BuildRight (71)
- : : : : :- * Filter (69)
- : : : : : +- * ColumnarToRow (68)
- : : : : : +- Scan parquet spark_catalog.default.catalog_sales (67)
- : : : : +- ReusedExchange (70)
- : : : +- ReusedExchange (72)
- : : +- ReusedExchange (75)
- : +- * Filter (96)
- : +- * HashAggregate (95)
- : +- Exchange (94)
- : +- * HashAggregate (93)
- : +- * Project (92)
- : +- * BroadcastHashJoin Inner BuildRight (91)
- : :- * Project (89)
- : : +- * BroadcastHashJoin Inner BuildRight (88)
- : : :- * BroadcastHashJoin LeftSemi BuildRight (86)
- : : : :- * Filter (84)
- : : : : +- * ColumnarToRow (83)
- : : : : +- Scan parquet spark_catalog.default.web_sales (82)
- : : : +- ReusedExchange (85)
- : : +- ReusedExchange (87)
- : +- ReusedExchange (90)
- :- * HashAggregate (105)
- : +- Exchange (104)
- : +- * HashAggregate (103)
- : +- * HashAggregate (102)
- : +- ReusedExchange (101)
- :- * HashAggregate (110)
- : +- Exchange (109)
- : +- * HashAggregate (108)
- : +- * HashAggregate (107)
- : +- ReusedExchange (106)
- :- * HashAggregate (115)
- : +- Exchange (114)
- : +- * HashAggregate (113)
- : +- * HashAggregate (112)
- : +- ReusedExchange (111)
- +- * HashAggregate (120)
- +- Exchange (119)
- +- * HashAggregate (118)
- +- * HashAggregate (117)
- +- ReusedExchange (116)
+TakeOrderedAndProject (164)
++- * HashAggregate (163)
+ +- Exchange (162)
+ +- * HashAggregate (161)
+ +- Union (160)
+ :- * HashAggregate (99)
+ : +- * HashAggregate (98)
+ : +- Union (97)
+ : :- * Filter (66)
+ : : +- * HashAggregate (65)
+ : : +- Exchange (64)
+ : : +- * HashAggregate (63)
+ : : +- * Project (62)
+ : : +- * BroadcastHashJoin Inner BuildRight (61)
+ : : :- * Project (59)
+ : : : +- * BroadcastHashJoin Inner BuildRight (58)
+ : : : :- * BroadcastHashJoin LeftSemi BuildRight (51)
+ : : : : :- * Filter (3)
+ : : : : : +- * ColumnarToRow (2)
+ : : : : : +- Scan parquet spark_catalog.default.store_sales (1)
+ : : : : +- BroadcastExchange (50)
+ : : : : +- * Project (49)
+ : : : : +- * BroadcastHashJoin Inner BuildRight (48)
+ : : : : :- * Filter (6)
+ : : : : : +- * ColumnarToRow (5)
+ : : : : : +- Scan parquet spark_catalog.default.item (4)
+ : : : : +- BroadcastExchange (47)
+ : : : : +- * BroadcastHashJoin LeftSemi BuildRight (46)
+ : : : : :- * HashAggregate (35)
+ : : : : : +- Exchange (34)
+ : : : : : +- * HashAggregate (33)
+ : : : : : +- * Project (32)
+ : : : : : +- * BroadcastHashJoin Inner BuildRight (31)
+ : : : : : :- * Project (29)
+ : : : : : : +- * BroadcastHashJoin Inner BuildRight (28)
+ : : : : : : :- * Filter (9)
+ : : : : : : : +- * ColumnarToRow (8)
+ : : : : : : : +- Scan parquet spark_catalog.default.store_sales (7)
+ : : : : : : +- BroadcastExchange (27)
+ : : : : : : +- * BroadcastHashJoin LeftSemi BuildRight (26)
+ : : : : : : :- * Filter (12)
+ : : : : : : : +- * ColumnarToRow (11)
+ : : : : : : : +- Scan parquet spark_catalog.default.item (10)
+ : : : : : : +- BroadcastExchange (25)
+ : : : : : : +- * Project (24)
+ : : : : : : +- * BroadcastHashJoin Inner BuildRight (23)
+ : : : : : : :- * Project (21)
+ : : : : : : : +- * BroadcastHashJoin Inner BuildRight (20)
+ : : : : : : : :- * Filter (15)
+ : : : : : : : : +- * ColumnarToRow (14)
+ : : : : : : : : +- Scan parquet spark_catalog.default.catalog_sales (13)
+ : : : : : : : +- BroadcastExchange (19)
+ : : : : : : : +- * Filter (18)
+ : : : : : : : +- * ColumnarToRow (17)
+ : : : : : : : +- Scan parquet spark_catalog.default.item (16)
+ : : : : : : +- ReusedExchange (22)
+ : : : : : +- ReusedExchange (30)
+ : : : : +- BroadcastExchange (45)
+ : : : : +- * Project (44)
+ : : : : +- * BroadcastHashJoin Inner BuildRight (43)
+ : : : : :- * Project (41)
+ : : : : : +- * BroadcastHashJoin Inner BuildRight (40)
+ : : : : : :- * Filter (38)
+ : : : : : : +- * ColumnarToRow (37)
+ : : : : : : +- Scan parquet spark_catalog.default.web_sales (36)
+ : : : : : +- ReusedExchange (39)
+ : : : : +- ReusedExchange (42)
+ : : : +- BroadcastExchange (57)
+ : : : +- * BroadcastHashJoin LeftSemi BuildRight (56)
+ : : : :- * Filter (54)
+ : : : : +- * ColumnarToRow (53)
+ : : : : +- Scan parquet spark_catalog.default.item (52)
+ : : : +- ReusedExchange (55)
+ : : +- ReusedExchange (60)
+ : :- * Filter (81)
+ : : +- * HashAggregate (80)
+ : : +- Exchange (79)
+ : : +- * HashAggregate (78)
+ : : +- * Project (77)
+ : : +- * BroadcastHashJoin Inner BuildRight (76)
+ : : :- * Project (74)
+ : : : +- * BroadcastHashJoin Inner BuildRight (73)
+ : : : :- * BroadcastHashJoin LeftSemi BuildRight (71)
+ : : : : :- * Filter (69)
+ : : : : : +- * ColumnarToRow (68)
+ : : : : : +- Scan parquet spark_catalog.default.catalog_sales (67)
+ : : : : +- ReusedExchange (70)
+ : : : +- ReusedExchange (72)
+ : : +- ReusedExchange (75)
+ : +- * Filter (96)
+ : +- * HashAggregate (95)
+ : +- Exchange (94)
+ : +- * HashAggregate (93)
+ : +- * Project (92)
+ : +- * BroadcastHashJoin Inner BuildRight (91)
+ : :- * Project (89)
+ : : +- * BroadcastHashJoin Inner BuildRight (88)
+ : : :- * BroadcastHashJoin LeftSemi BuildRight (86)
+ : : : :- * Filter (84)
+ : : : : +- * ColumnarToRow (83)
+ : : : : +- Scan parquet spark_catalog.default.web_sales (82)
+ : : : +- ReusedExchange (85)
+ : : +- ReusedExchange (87)
+ : +- ReusedExchange (90)
+ :- * HashAggregate (114)
+ : +- Exchange (113)
+ : +- * HashAggregate (112)
+ : +- * HashAggregate (111)
+ : +- * HashAggregate (110)
+ : +- Union (109)
Review Comment:
Ouch, this might cause performance regression as we no longer have `Exchange (99)` which we could reuse.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for same output partitoning as children operators [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2234131968
##########
sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q14a/explain.txt:
##########
@@ -1,129 +1,168 @@
== Physical Plan ==
-TakeOrderedAndProject (125)
-+- * HashAggregate (124)
- +- Exchange (123)
- +- * HashAggregate (122)
- +- Union (121)
- :- * HashAggregate (100)
- : +- Exchange (99)
- : +- * HashAggregate (98)
- : +- Union (97)
- : :- * Filter (66)
- : : +- * HashAggregate (65)
- : : +- Exchange (64)
- : : +- * HashAggregate (63)
- : : +- * Project (62)
- : : +- * BroadcastHashJoin Inner BuildRight (61)
- : : :- * Project (59)
- : : : +- * BroadcastHashJoin Inner BuildRight (58)
- : : : :- * BroadcastHashJoin LeftSemi BuildRight (51)
- : : : : :- * Filter (3)
- : : : : : +- * ColumnarToRow (2)
- : : : : : +- Scan parquet spark_catalog.default.store_sales (1)
- : : : : +- BroadcastExchange (50)
- : : : : +- * Project (49)
- : : : : +- * BroadcastHashJoin Inner BuildRight (48)
- : : : : :- * Filter (6)
- : : : : : +- * ColumnarToRow (5)
- : : : : : +- Scan parquet spark_catalog.default.item (4)
- : : : : +- BroadcastExchange (47)
- : : : : +- * BroadcastHashJoin LeftSemi BuildRight (46)
- : : : : :- * HashAggregate (35)
- : : : : : +- Exchange (34)
- : : : : : +- * HashAggregate (33)
- : : : : : +- * Project (32)
- : : : : : +- * BroadcastHashJoin Inner BuildRight (31)
- : : : : : :- * Project (29)
- : : : : : : +- * BroadcastHashJoin Inner BuildRight (28)
- : : : : : : :- * Filter (9)
- : : : : : : : +- * ColumnarToRow (8)
- : : : : : : : +- Scan parquet spark_catalog.default.store_sales (7)
- : : : : : : +- BroadcastExchange (27)
- : : : : : : +- * BroadcastHashJoin LeftSemi BuildRight (26)
- : : : : : : :- * Filter (12)
- : : : : : : : +- * ColumnarToRow (11)
- : : : : : : : +- Scan parquet spark_catalog.default.item (10)
- : : : : : : +- BroadcastExchange (25)
- : : : : : : +- * Project (24)
- : : : : : : +- * BroadcastHashJoin Inner BuildRight (23)
- : : : : : : :- * Project (21)
- : : : : : : : +- * BroadcastHashJoin Inner BuildRight (20)
- : : : : : : : :- * Filter (15)
- : : : : : : : : +- * ColumnarToRow (14)
- : : : : : : : : +- Scan parquet spark_catalog.default.catalog_sales (13)
- : : : : : : : +- BroadcastExchange (19)
- : : : : : : : +- * Filter (18)
- : : : : : : : +- * ColumnarToRow (17)
- : : : : : : : +- Scan parquet spark_catalog.default.item (16)
- : : : : : : +- ReusedExchange (22)
- : : : : : +- ReusedExchange (30)
- : : : : +- BroadcastExchange (45)
- : : : : +- * Project (44)
- : : : : +- * BroadcastHashJoin Inner BuildRight (43)
- : : : : :- * Project (41)
- : : : : : +- * BroadcastHashJoin Inner BuildRight (40)
- : : : : : :- * Filter (38)
- : : : : : : +- * ColumnarToRow (37)
- : : : : : : +- Scan parquet spark_catalog.default.web_sales (36)
- : : : : : +- ReusedExchange (39)
- : : : : +- ReusedExchange (42)
- : : : +- BroadcastExchange (57)
- : : : +- * BroadcastHashJoin LeftSemi BuildRight (56)
- : : : :- * Filter (54)
- : : : : +- * ColumnarToRow (53)
- : : : : +- Scan parquet spark_catalog.default.item (52)
- : : : +- ReusedExchange (55)
- : : +- ReusedExchange (60)
- : :- * Filter (81)
- : : +- * HashAggregate (80)
- : : +- Exchange (79)
- : : +- * HashAggregate (78)
- : : +- * Project (77)
- : : +- * BroadcastHashJoin Inner BuildRight (76)
- : : :- * Project (74)
- : : : +- * BroadcastHashJoin Inner BuildRight (73)
- : : : :- * BroadcastHashJoin LeftSemi BuildRight (71)
- : : : : :- * Filter (69)
- : : : : : +- * ColumnarToRow (68)
- : : : : : +- Scan parquet spark_catalog.default.catalog_sales (67)
- : : : : +- ReusedExchange (70)
- : : : +- ReusedExchange (72)
- : : +- ReusedExchange (75)
- : +- * Filter (96)
- : +- * HashAggregate (95)
- : +- Exchange (94)
- : +- * HashAggregate (93)
- : +- * Project (92)
- : +- * BroadcastHashJoin Inner BuildRight (91)
- : :- * Project (89)
- : : +- * BroadcastHashJoin Inner BuildRight (88)
- : : :- * BroadcastHashJoin LeftSemi BuildRight (86)
- : : : :- * Filter (84)
- : : : : +- * ColumnarToRow (83)
- : : : : +- Scan parquet spark_catalog.default.web_sales (82)
- : : : +- ReusedExchange (85)
- : : +- ReusedExchange (87)
- : +- ReusedExchange (90)
- :- * HashAggregate (105)
- : +- Exchange (104)
- : +- * HashAggregate (103)
- : +- * HashAggregate (102)
- : +- ReusedExchange (101)
- :- * HashAggregate (110)
- : +- Exchange (109)
- : +- * HashAggregate (108)
- : +- * HashAggregate (107)
- : +- ReusedExchange (106)
- :- * HashAggregate (115)
- : +- Exchange (114)
- : +- * HashAggregate (113)
- : +- * HashAggregate (112)
- : +- ReusedExchange (111)
- +- * HashAggregate (120)
- +- Exchange (119)
- +- * HashAggregate (118)
- +- * HashAggregate (117)
- +- ReusedExchange (116)
+TakeOrderedAndProject (164)
++- * HashAggregate (163)
+ +- Exchange (162)
+ +- * HashAggregate (161)
+ +- Union (160)
+ :- * HashAggregate (99)
+ : +- * HashAggregate (98)
+ : +- Union (97)
+ : :- * Filter (66)
+ : : +- * HashAggregate (65)
+ : : +- Exchange (64)
+ : : +- * HashAggregate (63)
+ : : +- * Project (62)
+ : : +- * BroadcastHashJoin Inner BuildRight (61)
+ : : :- * Project (59)
+ : : : +- * BroadcastHashJoin Inner BuildRight (58)
+ : : : :- * BroadcastHashJoin LeftSemi BuildRight (51)
+ : : : : :- * Filter (3)
+ : : : : : +- * ColumnarToRow (2)
+ : : : : : +- Scan parquet spark_catalog.default.store_sales (1)
+ : : : : +- BroadcastExchange (50)
+ : : : : +- * Project (49)
+ : : : : +- * BroadcastHashJoin Inner BuildRight (48)
+ : : : : :- * Filter (6)
+ : : : : : +- * ColumnarToRow (5)
+ : : : : : +- Scan parquet spark_catalog.default.item (4)
+ : : : : +- BroadcastExchange (47)
+ : : : : +- * BroadcastHashJoin LeftSemi BuildRight (46)
+ : : : : :- * HashAggregate (35)
+ : : : : : +- Exchange (34)
+ : : : : : +- * HashAggregate (33)
+ : : : : : +- * Project (32)
+ : : : : : +- * BroadcastHashJoin Inner BuildRight (31)
+ : : : : : :- * Project (29)
+ : : : : : : +- * BroadcastHashJoin Inner BuildRight (28)
+ : : : : : : :- * Filter (9)
+ : : : : : : : +- * ColumnarToRow (8)
+ : : : : : : : +- Scan parquet spark_catalog.default.store_sales (7)
+ : : : : : : +- BroadcastExchange (27)
+ : : : : : : +- * BroadcastHashJoin LeftSemi BuildRight (26)
+ : : : : : : :- * Filter (12)
+ : : : : : : : +- * ColumnarToRow (11)
+ : : : : : : : +- Scan parquet spark_catalog.default.item (10)
+ : : : : : : +- BroadcastExchange (25)
+ : : : : : : +- * Project (24)
+ : : : : : : +- * BroadcastHashJoin Inner BuildRight (23)
+ : : : : : : :- * Project (21)
+ : : : : : : : +- * BroadcastHashJoin Inner BuildRight (20)
+ : : : : : : : :- * Filter (15)
+ : : : : : : : : +- * ColumnarToRow (14)
+ : : : : : : : : +- Scan parquet spark_catalog.default.catalog_sales (13)
+ : : : : : : : +- BroadcastExchange (19)
+ : : : : : : : +- * Filter (18)
+ : : : : : : : +- * ColumnarToRow (17)
+ : : : : : : : +- Scan parquet spark_catalog.default.item (16)
+ : : : : : : +- ReusedExchange (22)
+ : : : : : +- ReusedExchange (30)
+ : : : : +- BroadcastExchange (45)
+ : : : : +- * Project (44)
+ : : : : +- * BroadcastHashJoin Inner BuildRight (43)
+ : : : : :- * Project (41)
+ : : : : : +- * BroadcastHashJoin Inner BuildRight (40)
+ : : : : : :- * Filter (38)
+ : : : : : : +- * ColumnarToRow (37)
+ : : : : : : +- Scan parquet spark_catalog.default.web_sales (36)
+ : : : : : +- ReusedExchange (39)
+ : : : : +- ReusedExchange (42)
+ : : : +- BroadcastExchange (57)
+ : : : +- * BroadcastHashJoin LeftSemi BuildRight (56)
+ : : : :- * Filter (54)
+ : : : : +- * ColumnarToRow (53)
+ : : : : +- Scan parquet spark_catalog.default.item (52)
+ : : : +- ReusedExchange (55)
+ : : +- ReusedExchange (60)
+ : :- * Filter (81)
+ : : +- * HashAggregate (80)
+ : : +- Exchange (79)
+ : : +- * HashAggregate (78)
+ : : +- * Project (77)
+ : : +- * BroadcastHashJoin Inner BuildRight (76)
+ : : :- * Project (74)
+ : : : +- * BroadcastHashJoin Inner BuildRight (73)
+ : : : :- * BroadcastHashJoin LeftSemi BuildRight (71)
+ : : : : :- * Filter (69)
+ : : : : : +- * ColumnarToRow (68)
+ : : : : : +- Scan parquet spark_catalog.default.catalog_sales (67)
+ : : : : +- ReusedExchange (70)
+ : : : +- ReusedExchange (72)
+ : : +- ReusedExchange (75)
+ : +- * Filter (96)
+ : +- * HashAggregate (95)
+ : +- Exchange (94)
+ : +- * HashAggregate (93)
+ : +- * Project (92)
+ : +- * BroadcastHashJoin Inner BuildRight (91)
+ : :- * Project (89)
+ : : +- * BroadcastHashJoin Inner BuildRight (88)
+ : : :- * BroadcastHashJoin LeftSemi BuildRight (86)
+ : : : :- * Filter (84)
+ : : : : +- * ColumnarToRow (83)
+ : : : : +- Scan parquet spark_catalog.default.web_sales (82)
+ : : : +- ReusedExchange (85)
+ : : +- ReusedExchange (87)
+ : +- ReusedExchange (90)
+ :- * HashAggregate (114)
+ : +- Exchange (113)
+ : +- * HashAggregate (112)
+ : +- * HashAggregate (111)
+ : +- * HashAggregate (110)
+ : +- Union (109)
Review Comment:
Yeah. Unfortunately the above happens 4 times in this query.
But still, IMO this PR is a step in the right direction to remove unnecessary exchanges.
In the future some kind of common subplan elimination rule should identify and cache query parts that occur multiple times if computation costs exceed the cost of caching.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2232821528
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,8 +701,42 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
Review Comment:
Actaully, does this mean that children executions are triggered to get `outputPartitioning` of an `Union`?
E.g. a simple explain to show the physical plan can now trigger execuion of union children?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3119862901
Could you re-trigger `ThriftServer` tests?
```
[info] *** 3 TESTS FAILED ***
[error] Failed tests:
[error] org.apache.spark.sql.hive.thriftserver.ThriftServerQueryTestSuite
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2232606389
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala:
##########
@@ -145,6 +145,12 @@ case class BroadcastExchangeExec(
Statistics(dataSize, Some(rowCount))
}
+ override def resetMetrics(): Unit = {
+ // no-op
+ // BroadcastExchangeExec after materialized won't be materialized again, so we should not
+ // reset the metrics. Otherwise, we will lose the metrics collected in the broadcast job.
+ }
Review Comment:
https://github.com/apache/spark/pull/51673 also has related test case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2233105944
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,8 +701,42 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
Review Comment:
I actually am working the next PR locally. After this gets merged, I will open a new PR to improve it and get rid of this `execute` calls.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52921][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3119966934
> Could you re-trigger `ThriftServer` tests?
>
> ```
> [info] *** 3 TESTS FAILED ***
> [error] Failed tests:
> [error] org.apache.spark.sql.hive.thriftserver.ThriftServerQueryTestSuite
> ```
They are related test failures. I'm investigating them. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-XXXXX][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #51623:
URL: https://github.com/apache/spark/pull/51623#discussion_r2223895260
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -699,6 +699,18 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
}
}
+ private lazy val childrenRDDs = children.map(_.execute())
+
+ override def outputPartitioning: Partitioning = {
+ val nonEmptyRdds = childrenRDDs.filter(!_.partitions.isEmpty)
Review Comment:
Because `SparkContext.union` uses `nonEmptyRdds`, so I didn't move `nonEmptyRdds` logic into `isPartitionerAwareUnion`. I leave to the callers to pass in non empty rdds.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-XXXXX][SQL] Specify outputPartitioning for UnionExec for partitioner aware case [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #51623:
URL: https://github.com/apache/spark/pull/51623#issuecomment-3104987519
Note that this is marked as draft as I don't add test yet. Will add test later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org