You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2020/08/19 01:14:24 UTC
[GitHub] [spark] imback82 opened a new pull request #29473: [WIP][SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
imback82 opened a new pull request #29473:
URL: https://github.com/apache/spark/pull/29473
<!--
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'.
-->
### 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.
-->
#28123 and #29079 introduced coalescing bucketed tables for sort merge join / shuffled hash join.
This PR proposes to introduce repartitioning bucketed tables to increase parallelism at the cost of reading duplicate source data. It is applied if the following conditions are met:
* Join is sort merge join or shuffled hash join.
* Join keys match with output partition expressions on their respective sides.
* The larger bucket count is divisible by the smaller bucket count.
* `spark.sql.bucketing.repartitionBucketsInJoin.enabled` is set to "true".
* The ratio of the number of buckets should be less than the value set in `spark.sql.bucketing.coalesceOrRepartitionBucketsInJoin.maxBucketRatio`.
### 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.
-->
Coalescing buckets is useful but repartitioning can also help due to the increased parallelism depending on the workloads.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
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'.
-->
Yes. If the bucket repartitioning conditions explained above are met, a full shuffle can be eliminated (also note that you will see `SelectedBucketsCount: 4 out of 4 (Repartitioned to 8)` in the physical plan):
```
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "0")
spark.conf.set("spark.sql.bucketing.repartitionBucketsInJoin.enabled", "true")
val df1 = (0 until 20).map(i => (i % 5, i % 13, i.toString)).toDF("i", "j", "k")
val df2 = (0 until 20).map(i => (i % 7, i % 11, i.toString)).toDF("i", "j", "k")
df1.write.format("parquet").bucketBy(8, "i").saveAsTable("t1")
df2.write.format("parquet").bucketBy(4, "i").saveAsTable("t2")
val t1 = spark.table("t1")
val t2 = spark.table("t2")
val joined = t1.join(t2, t1("i") === t2("i"))
joined.explain
== Physical Plan ==
*(3) SortMergeJoin [i#38], [i#44], Inner
:- *(1) Sort [i#38 ASC NULLS FIRST], false, 0
: +- *(1) Filter isnotnull(i#38)
: +- *(1) ColumnarToRow
: +- FileScan parquet default.t1[i#38,j#39,k#40] Batched: true, DataFilters: [isnotnull(i#38)], Format: Parquet, Location: InMemoryFileIndex[file:/Users/terryk/repos/GitHub/imback82/apache-spark/spark-warehouse/t1], PartitionFilters: [], PushedFilters: [IsNotNull(i)], ReadSchema: struct<i:int,j:int,k:string>, SelectedBucketsCount: 8 out of 8
+- *(2) Sort [i#44 ASC NULLS FIRST], false, 0
+- *(2) Filter isnotnull(i#44)
+- FileScan parquet default.t2[i#44,j#45,k#46] Batched: false, DataFilters: [isnotnull(i#44)], Format: Parquet, Location: InMemoryFileIndex[file:/Users/terryk/repos/GitHub/imback82/apache-spark/spark-warehouse/t2], PartitionFilters: [], PushedFilters: [IsNotNull(i)], ReadSchema: struct<i:int,j:int,k:string>, SelectedBucketsCount: 4 out of 4 (Repartitioned to 8)
```
### 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.
-->
Added new 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.
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
[GitHub] [spark] imback82 commented on a change in pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
imback82 commented on a change in pull request #29473:
URL: https://github.com/apache/spark/pull/29473#discussion_r478594277
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -166,15 +167,24 @@ case class FileSourceScanExec(
requiredSchema: StructType,
partitionFilters: Seq[Expression],
optionalBucketSet: Option[BitSet],
- optionalNumCoalescedBuckets: Option[Int],
+ optionalNewNumBuckets: Option[Int],
dataFilters: Seq[Expression],
tableIdentifier: Option[TableIdentifier])
extends DataSourceScanExec {
// Note that some vals referring the file-based relation are lazy intentionally
// so that this plan can be canonicalized on executor side too. See SPARK-23731.
override lazy val supportsColumnar: Boolean = {
- relation.fileFormat.supportBatch(relation.sparkSession, schema)
+ // `RepartitioningBucketRDD` converts columnar batches to rows to calculate bucket id for each
+ // row, thus columnar is not supported when `RepartitioningBucketRDD` is used to avoid
+ // conversions from batches to rows and back to batches.
+ relation.fileFormat.supportBatch(relation.sparkSession, schema) && !isRepartitioningBuckets
Review comment:
Note that the datasource will still be read as batches in this case (if whole stage codegen is enabled).
I see that physical plans operate on rows, so batches are converted to rows via `ColumnarToRow` anyway. So, I think perf impact would be minimal here; the difference could be the code-gen conversion from columnar to row vs. iterating `batch.rowIterator()` in `BucketRepartitioningRDD`.
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-675797802
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-675876095
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683375706
**[Test build #128028 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128028/testReport)** for PR 29473 at commit [`366c9c3`](https://github.com/apache/spark/commit/366c9c3b9a8872113d7ccf670b3e8bfbb53f892e).
----------------------------------------------------------------
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.
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
[GitHub] [spark] imback82 commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
imback82 commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-682293826
> btw, still `Draft`?
I still need to add few more tests.
> Also, I think its better to describe performance numbers for this proposed idea in the PR description above.
Yes, will update.
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683375765
----------------------------------------------------------------
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.
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
[GitHub] [spark] imback82 commented on a change in pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
imback82 commented on a change in pull request #29473:
URL: https://github.com/apache/spark/pull/29473#discussion_r478594277
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -166,15 +167,24 @@ case class FileSourceScanExec(
requiredSchema: StructType,
partitionFilters: Seq[Expression],
optionalBucketSet: Option[BitSet],
- optionalNumCoalescedBuckets: Option[Int],
+ optionalNewNumBuckets: Option[Int],
dataFilters: Seq[Expression],
tableIdentifier: Option[TableIdentifier])
extends DataSourceScanExec {
// Note that some vals referring the file-based relation are lazy intentionally
// so that this plan can be canonicalized on executor side too. See SPARK-23731.
override lazy val supportsColumnar: Boolean = {
- relation.fileFormat.supportBatch(relation.sparkSession, schema)
+ // `RepartitioningBucketRDD` converts columnar batches to rows to calculate bucket id for each
+ // row, thus columnar is not supported when `RepartitioningBucketRDD` is used to avoid
+ // conversions from batches to rows and back to batches.
+ relation.fileFormat.supportBatch(relation.sparkSession, schema) && !isRepartitioningBuckets
Review comment:
Note that the datasource will still be read as batches in this case (if whole stage codegen is enabled).
I see that physical plans operates on rows, so batches are converted to rows via `ColumnarToRow` anyway. So, I think perf impact would be minimal here; the difference could be the code-gen conversion from columnar to row vs. iterating `batch.rowIterator()` in `BucketRepartitioningRDD`.
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683375370
----------------------------------------------------------------
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.
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
[GitHub] [spark] github-actions[bot] closed pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #29473:
URL: https://github.com/apache/spark/pull/29473
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-675797404
**[Test build #127609 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127609/testReport)** for PR 29473 at commit [`21882ab`](https://github.com/apache/spark/commit/21882abe495ce6c1fff297ec7d52928f13a2fcc9).
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683211632
**[Test build #128011 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128011/testReport)** for PR 29473 at commit [`e2374ac`](https://github.com/apache/spark/commit/e2374ac281bbcb23c0dc49786ce7d8148f9761bd).
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683385078
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683241185
**[Test build #128011 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128011/testReport)** for PR 29473 at commit [`e2374ac`](https://github.com/apache/spark/commit/e2374ac281bbcb23c0dc49786ce7d8148f9761bd).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683241374
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683211632
**[Test build #128011 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128011/testReport)** for PR 29473 at commit [`e2374ac`](https://github.com/apache/spark/commit/e2374ac281bbcb23c0dc49786ce7d8148f9761bd).
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683219778
----------------------------------------------------------------
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.
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
[GitHub] [spark] maropu commented on a change in pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #29473:
URL: https://github.com/apache/spark/pull/29473#discussion_r475587292
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -166,15 +167,24 @@ case class FileSourceScanExec(
requiredSchema: StructType,
partitionFilters: Seq[Expression],
optionalBucketSet: Option[BitSet],
- optionalNumCoalescedBuckets: Option[Int],
+ optionalNewNumBuckets: Option[Int],
dataFilters: Seq[Expression],
tableIdentifier: Option[TableIdentifier])
extends DataSourceScanExec {
// Note that some vals referring the file-based relation are lazy intentionally
// so that this plan can be canonicalized on executor side too. See SPARK-23731.
override lazy val supportsColumnar: Boolean = {
- relation.fileFormat.supportBatch(relation.sparkSession, schema)
+ // `RepartitioningBucketRDD` converts columnar batches to rows to calculate bucket id for each
+ // row, thus columnar is not supported when `RepartitioningBucketRDD` is used to avoid
+ // conversions from batches to rows and back to batches.
+ relation.fileFormat.supportBatch(relation.sparkSession, schema) && !isRepartitioningBuckets
Review comment:
I'm not sure about how much this columnar execution makes performance gains though, the proposed idea is to give up the gains then use bucket repartitioning instead?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/bucketing/CoalesceOrRepartitionBucketsInJoin.scala
##########
@@ -83,23 +82,39 @@ case class CoalesceBucketsInJoin(conf: SQLConf) extends Rule[SparkPlan] {
}
def apply(plan: SparkPlan): SparkPlan = {
- if (!conf.coalesceBucketsInJoinEnabled) {
+ if (!conf.coalesceBucketsInJoinEnabled && !conf.repartitionBucketsInJoinEnabled) {
return plan
}
+ if (conf.coalesceBucketsInJoinEnabled && conf.repartitionBucketsInJoinEnabled) {
+ throw new AnalysisException("Both 'spark.sql.bucketing.coalesceBucketsInJoin.enabled' and " +
+ "'spark.sql.bucketing.repartitionBucketsInJoin.enabled' cannot be set to true at the" +
+ "same time")
Review comment:
Could we use `Enumeration` and `checkValues` instead? I think this check should be done in `SQLConf`.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -166,15 +167,24 @@ case class FileSourceScanExec(
requiredSchema: StructType,
partitionFilters: Seq[Expression],
optionalBucketSet: Option[BitSet],
- optionalNumCoalescedBuckets: Option[Int],
+ optionalNewNumBuckets: Option[Int],
dataFilters: Seq[Expression],
tableIdentifier: Option[TableIdentifier])
extends DataSourceScanExec {
// Note that some vals referring the file-based relation are lazy intentionally
// so that this plan can be canonicalized on executor side too. See SPARK-23731.
override lazy val supportsColumnar: Boolean = {
- relation.fileFormat.supportBatch(relation.sparkSession, schema)
+ // `RepartitioningBucketRDD` converts columnar batches to rows to calculate bucket id for each
+ // row, thus columnar is not supported when `RepartitioningBucketRDD` is used to avoid
+ // conversions from batches to rows and back to batches.
+ relation.fileFormat.supportBatch(relation.sparkSession, schema) && !isRepartitioningBuckets
+ }
+
+ @transient private lazy val isRepartitioningBuckets: Boolean = {
Review comment:
nit: we don't need `: Boolean `?
----------------------------------------------------------------
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.
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
[GitHub] [spark] maropu commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
maropu commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-679105200
Thanks for the work, @imback82. Just a question; we cannot set a simpler rule to select which strategy (reapportioning or coalescing) we use when reading buckets? I think it is a bit annoying to set true to `repartitionBucketsInJoin.enabled` or `coalesceBucketsInJoin.enabled` data-by-data. What's a factor to make a difference between the two strategies, e.g., bucket size? For example, if one has too small buckets (this case is extreme though), repartitioning for higher cardinality might not be able to help for better performance.
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683227452
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-675875114
**[Test build #127609 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127609/testReport)** for PR 29473 at commit [`21882ab`](https://github.com/apache/spark/commit/21882abe495ce6c1fff297ec7d52928f13a2fcc9).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683227251
**[Test build #128012 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128012/testReport)** for PR 29473 at commit [`7481e36`](https://github.com/apache/spark/commit/7481e36d8781e869a0dc558e0af5d358a56ab150).
----------------------------------------------------------------
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.
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
[GitHub] [spark] c21 commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683245934
> did you observe any patterns or heuristics on your workloads where repartition is preferred?
From our side, honestly now we don't have any automation for deciding coalesce vs repartition. We provided configs similar here for users themselves to control coalesce vs repartition.
I think a rule of thumb can be we don't want to
(1).coalesce: if the coalesced table is too big and # of coalesced buckets is too few, then each task has too much data and will take more time.
(2).repartition: if the repartition table is too big and # of repartitioned buckets is too many, then too much duplicated data is read and will have too much more CPU/IO cost (might be worse than just shuffling this table).
----------------------------------------------------------------
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.
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
[GitHub] [spark] imback82 commented on a change in pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
imback82 commented on a change in pull request #29473:
URL: https://github.com/apache/spark/pull/29473#discussion_r479720956
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/bucketing/CoalesceOrRepartitionBucketsInJoin.scala
##########
@@ -27,45 +27,48 @@ import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.{FileSourceScanExec, FilterExec, ProjectExec, SparkPlan}
import org.apache.spark.sql.execution.joins.{BaseJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.SQLConf.BucketReadStrategyInJoin
/**
- * This rule coalesces one side of the `SortMergeJoin` and `ShuffledHashJoin`
+ * This rule coalesces or repartitions one side of the `SortMergeJoin` and `ShuffledHashJoin`
* if the following conditions are met:
* - Two bucketed tables are joined.
* - Join keys match with output partition expressions on their respective sides.
* - The larger bucket number is divisible by the smaller bucket number.
- * - COALESCE_BUCKETS_IN_JOIN_ENABLED is set to true.
* - The ratio of the number of buckets is less than the value set in
- * COALESCE_BUCKETS_IN_JOIN_MAX_BUCKET_RATIO.
+ * COALESCE_OR_REPARTITION_BUCKETS_IN_JOIN_MAX_BUCKET_RATIO.
Review comment:
Thanks for the catch.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -548,22 +560,42 @@ case class FileSourceScanExec(
filesGroupedToBuckets
}
- val filePartitions = optionalNumCoalescedBuckets.map { numCoalescedBuckets =>
- logInfo(s"Coalescing to ${numCoalescedBuckets} buckets")
- val coalescedBuckets = prunedFilesGroupedToBuckets.groupBy(_._1 % numCoalescedBuckets)
- Seq.tabulate(numCoalescedBuckets) { bucketId =>
- val partitionedFiles = coalescedBuckets.get(bucketId).map {
- _.values.flatten.toArray
- }.getOrElse(Array.empty)
- FilePartition(bucketId, partitionedFiles)
- }
- }.getOrElse {
- Seq.tabulate(bucketSpec.numBuckets) { bucketId =>
+ if (optionalNewNumBuckets.isEmpty) {
+ val filePartitions = Seq.tabulate(bucketSpec.numBuckets) { bucketId =>
FilePartition(bucketId, prunedFilesGroupedToBuckets.getOrElse(bucketId, Array.empty))
}
+ new FileScanRDD(fsRelation.sparkSession, readFile, filePartitions)
+ } else {
+ val newNumBuckets = optionalNewNumBuckets.get
+ if (newNumBuckets < bucketSpec.numBuckets) {
+ assert(bucketSpec.numBuckets % newNumBuckets == 0)
+ logInfo(s"Coalescing to $newNumBuckets buckets from ${bucketSpec.numBuckets} buckets")
+ val coalescedBuckets = prunedFilesGroupedToBuckets.groupBy(_._1 % newNumBuckets)
+ val filePartitions = Seq.tabulate(newNumBuckets) { bucketId =>
+ val partitionedFiles = coalescedBuckets
+ .get(bucketId)
+ .map(_.values.flatten.toArray)
+ .getOrElse(Array.empty)
+ FilePartition(bucketId, partitionedFiles)
+ }
+ new FileScanRDD(fsRelation.sparkSession, readFile, filePartitions)
+ } else {
+ assert(newNumBuckets % bucketSpec.numBuckets == 0)
+ logInfo(s"Repartitioning to $newNumBuckets buckets from ${bucketSpec.numBuckets} buckets")
+ val filePartitions = Seq.tabulate(newNumBuckets) { bucketId =>
+ FilePartition(
+ bucketId,
+ prunedFilesGroupedToBuckets.getOrElse(bucketId % bucketSpec.numBuckets, Array.empty))
+ }
+ // There are now more files to be read.
+ val filesNum = filePartitions.map(_.files.size.toLong).sum
+ val filesSize = filePartitions.map(_.files.map(_.length).sum).sum
+ driverMetrics("numFiles") = filesNum
+ driverMetrics("filesSize") = filesSize
+ new BucketRepartitioningRDD(
+ fsRelation.sparkSession, readFile, filePartitions, bucketSpec, newNumBuckets, output)
Review comment:
Thanks, updated.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -548,22 +560,42 @@ case class FileSourceScanExec(
filesGroupedToBuckets
}
- val filePartitions = optionalNumCoalescedBuckets.map { numCoalescedBuckets =>
- logInfo(s"Coalescing to ${numCoalescedBuckets} buckets")
- val coalescedBuckets = prunedFilesGroupedToBuckets.groupBy(_._1 % numCoalescedBuckets)
- Seq.tabulate(numCoalescedBuckets) { bucketId =>
- val partitionedFiles = coalescedBuckets.get(bucketId).map {
- _.values.flatten.toArray
- }.getOrElse(Array.empty)
- FilePartition(bucketId, partitionedFiles)
- }
- }.getOrElse {
- Seq.tabulate(bucketSpec.numBuckets) { bucketId =>
+ if (optionalNewNumBuckets.isEmpty) {
+ val filePartitions = Seq.tabulate(bucketSpec.numBuckets) { bucketId =>
FilePartition(bucketId, prunedFilesGroupedToBuckets.getOrElse(bucketId, Array.empty))
}
+ new FileScanRDD(fsRelation.sparkSession, readFile, filePartitions)
+ } else {
+ val newNumBuckets = optionalNewNumBuckets.get
+ if (newNumBuckets < bucketSpec.numBuckets) {
+ assert(bucketSpec.numBuckets % newNumBuckets == 0)
+ logInfo(s"Coalescing to $newNumBuckets buckets from ${bucketSpec.numBuckets} buckets")
+ val coalescedBuckets = prunedFilesGroupedToBuckets.groupBy(_._1 % newNumBuckets)
+ val filePartitions = Seq.tabulate(newNumBuckets) { bucketId =>
+ val partitionedFiles = coalescedBuckets
+ .get(bucketId)
+ .map(_.values.flatten.toArray)
+ .getOrElse(Array.empty)
+ FilePartition(bucketId, partitionedFiles)
+ }
+ new FileScanRDD(fsRelation.sparkSession, readFile, filePartitions)
+ } else {
+ assert(newNumBuckets % bucketSpec.numBuckets == 0)
+ logInfo(s"Repartitioning to $newNumBuckets buckets from ${bucketSpec.numBuckets} buckets")
+ val filePartitions = Seq.tabulate(newNumBuckets) { bucketId =>
+ FilePartition(
+ bucketId,
+ prunedFilesGroupedToBuckets.getOrElse(bucketId % bucketSpec.numBuckets, Array.empty))
+ }
+ // There are now more files to be read.
+ val filesNum = filePartitions.map(_.files.size.toLong).sum
+ val filesSize = filePartitions.map(_.files.map(_.length).sum).sum
+ driverMetrics("numFiles") = filesNum
Review comment:
I think `staticFilesNum` is used only for dynamic partition pruning:
https://github.com/apache/spark/blob/cfe012a4311a8fb1fc3a82390c3e68f6afcb1da6/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala#L421-L424
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -2655,24 +2655,34 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
- val COALESCE_BUCKETS_IN_JOIN_ENABLED =
- buildConf("spark.sql.bucketing.coalesceBucketsInJoin.enabled")
- .doc("When true, if two bucketed tables with the different number of buckets are joined, " +
- "the side with a bigger number of buckets will be coalesced to have the same number " +
- "of buckets as the other side. Bigger number of buckets is divisible by the smaller " +
- "number of buckets. Bucket coalescing is applied to sort-merge joins and " +
- "shuffled hash join. Note: Coalescing bucketed table can avoid unnecessary shuffling " +
- "in join, but it also reduces parallelism and could possibly cause OOM for " +
- "shuffled hash join.")
- .version("3.1.0")
- .booleanConf
- .createWithDefault(false)
+ object BucketReadStrategyInJoin extends Enumeration {
+ val COALESCE, REPARTITION, OFF = Value
+ }
- val COALESCE_BUCKETS_IN_JOIN_MAX_BUCKET_RATIO =
- buildConf("spark.sql.bucketing.coalesceBucketsInJoin.maxBucketRatio")
- .doc("The ratio of the number of two buckets being coalesced should be less than or " +
- "equal to this value for bucket coalescing to be applied. This configuration only " +
- s"has an effect when '${COALESCE_BUCKETS_IN_JOIN_ENABLED.key}' is set to true.")
+ val BUCKET_READ_STRATEGY_IN_JOIN =
+ buildConf("spark.sql.bucketing.bucketReadStrategyInJoin")
+ .doc("When set to COALESCE, if two bucketed tables with the different number of buckets " +
+ "are joined, the side with a bigger number of buckets will be coalesced to have the same " +
+ "number of buckets as the other side. When set to REPARTITION, the side with a bigger " +
+ "number of buckets will be repartitioned to have the same number of buckets as the other " +
+ "side. The bigger number of buckets must be divisible by the smaller number of buckets, " +
+ "and the strategy is applied to sort-merge joins and shuffled hash joins. " +
+ "Note: Coalescing bucketed table can avoid unnecessary shuffle in join, but it also " +
+ "reduces parallelism and could possibly cause OOM for shuffled hash join. Repartitioning " +
+ "bucketed table avoids unnecessary shuffle in join while maintaining the parallelism " +
+ "at the cost of reading duplicate data.")
+ .version("3.1.0")
+ .stringConf
+ .transform(_.toUpperCase(Locale.ROOT))
+ .checkValues(BucketReadStrategyInJoin.values.map(_.toString))
+ .createWithDefault(BucketReadStrategyInJoin.OFF.toString)
+
+ val BUCKET_READ_STRATEGY_IN_JOIN_MAX_BUCKET_RATIO =
+ buildConf("spark.sql.bucketing.bucketReadStrategyInJoin.maxBucketRatio")
Review comment:
I changed this to `spark.sql.sources.bucketing.readStrategyInJoin.maxBucketRatio`, but I don't have a strong opinion on this if `spark.sql.sources.bucketing.readStrategyInJoinMaxBucketRatio` is better.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/bucketing/BucketRepartitioningRDD.scala
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.bucketing
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.{Partition, TaskContext}
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.catalog.BucketSpec
+import org.apache.spark.sql.catalyst.expressions.{Attribute, UnsafeProjection}
+import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.execution.datasources.{FilePartition, FileScanRDD, PartitionedFile}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+
+/**
+ * An RDD that filters out the rows that do not belong to the current bucket file being read.
+ */
+private[spark] class BucketRepartitioningRDD(
+ @transient private val sparkSession: SparkSession,
+ readFunction: PartitionedFile => Iterator[InternalRow],
+ @transient override val filePartitions: Seq[FilePartition],
+ bucketSpec: BucketSpec,
+ numRepartitionedBuckets: Int,
+ output: Seq[Attribute])
+ extends FileScanRDD(sparkSession, readFunction, filePartitions) {
+ assert(numRepartitionedBuckets > bucketSpec.numBuckets)
Review comment:
I prefer having separate asserts when testing against multiple expressions (o.w. you don't know which assert failed since you only get the line number of assertion). Anyway, after [this](https://github.com/apache/spark/pull/29473/files#r479614691), the asserts are no longer needed.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -164,15 +165,24 @@ case class FileSourceScanExec(
requiredSchema: StructType,
partitionFilters: Seq[Expression],
optionalBucketSet: Option[BitSet],
- optionalNumCoalescedBuckets: Option[Int],
+ optionalNewNumBuckets: Option[Int],
dataFilters: Seq[Expression],
tableIdentifier: Option[TableIdentifier])
extends DataSourceScanExec {
// Note that some vals referring the file-based relation are lazy intentionally
// so that this plan can be canonicalized on executor side too. See SPARK-23731.
override lazy val supportsColumnar: Boolean = {
- relation.fileFormat.supportBatch(relation.sparkSession, schema)
+ // `RepartitioningBucketRDD` converts columnar batches to rows to calculate bucket id for each
+ // row, thus columnar is not supported when `RepartitioningBucketRDD` is used to avoid
+ // conversions from batches to rows and back to batches.
+ relation.fileFormat.supportBatch(relation.sparkSession, schema) && !isRepartitioningBuckets
+ }
+
+ @transient private lazy val isRepartitioningBuckets: Boolean = {
+ relation.bucketSpec.isDefined &&
+ optionalNewNumBuckets.isDefined &&
+ optionalNewNumBuckets.get > relation.bucketSpec.get.numBuckets
Review comment:
Yes, changed this as suggeted.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -314,7 +324,7 @@ case class FileSourceScanExec(
val singleFilePartitions = bucketToFilesGrouping.forall(p => p._2.length <= 1)
// TODO SPARK-24528 Sort order is currently ignored if buckets are coalesced.
- if (singleFilePartitions && optionalNumCoalescedBuckets.isEmpty) {
+ if (singleFilePartitions && (optionalNewNumBuckets.isEmpty || isRepartitioningBuckets)) {
Review comment:
Repartition can still maintain the sort order whereas coalescing cannot, thus this check is needed.
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -2655,24 +2655,34 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
- val COALESCE_BUCKETS_IN_JOIN_ENABLED =
- buildConf("spark.sql.bucketing.coalesceBucketsInJoin.enabled")
- .doc("When true, if two bucketed tables with the different number of buckets are joined, " +
- "the side with a bigger number of buckets will be coalesced to have the same number " +
- "of buckets as the other side. Bigger number of buckets is divisible by the smaller " +
- "number of buckets. Bucket coalescing is applied to sort-merge joins and " +
- "shuffled hash join. Note: Coalescing bucketed table can avoid unnecessary shuffling " +
- "in join, but it also reduces parallelism and could possibly cause OOM for " +
- "shuffled hash join.")
- .version("3.1.0")
- .booleanConf
- .createWithDefault(false)
+ object BucketReadStrategyInJoin extends Enumeration {
+ val COALESCE, REPARTITION, OFF = Value
+ }
- val COALESCE_BUCKETS_IN_JOIN_MAX_BUCKET_RATIO =
- buildConf("spark.sql.bucketing.coalesceBucketsInJoin.maxBucketRatio")
- .doc("The ratio of the number of two buckets being coalesced should be less than or " +
- "equal to this value for bucket coalescing to be applied. This configuration only " +
- s"has an effect when '${COALESCE_BUCKETS_IN_JOIN_ENABLED.key}' is set to true.")
+ val BUCKET_READ_STRATEGY_IN_JOIN =
+ buildConf("spark.sql.bucketing.bucketReadStrategyInJoin")
Review comment:
Makes sense. I change this to `spark.sql.sources.bucketing.readStrategyInJoin`.
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -2655,24 +2655,34 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
- val COALESCE_BUCKETS_IN_JOIN_ENABLED =
- buildConf("spark.sql.bucketing.coalesceBucketsInJoin.enabled")
- .doc("When true, if two bucketed tables with the different number of buckets are joined, " +
- "the side with a bigger number of buckets will be coalesced to have the same number " +
- "of buckets as the other side. Bigger number of buckets is divisible by the smaller " +
- "number of buckets. Bucket coalescing is applied to sort-merge joins and " +
- "shuffled hash join. Note: Coalescing bucketed table can avoid unnecessary shuffling " +
- "in join, but it also reduces parallelism and could possibly cause OOM for " +
- "shuffled hash join.")
- .version("3.1.0")
- .booleanConf
- .createWithDefault(false)
+ object BucketReadStrategyInJoin extends Enumeration {
+ val COALESCE, REPARTITION, OFF = Value
+ }
- val COALESCE_BUCKETS_IN_JOIN_MAX_BUCKET_RATIO =
- buildConf("spark.sql.bucketing.coalesceBucketsInJoin.maxBucketRatio")
- .doc("The ratio of the number of two buckets being coalesced should be less than or " +
- "equal to this value for bucket coalescing to be applied. This configuration only " +
- s"has an effect when '${COALESCE_BUCKETS_IN_JOIN_ENABLED.key}' is set to true.")
+ val BUCKET_READ_STRATEGY_IN_JOIN =
+ buildConf("spark.sql.bucketing.bucketReadStrategyInJoin")
+ .doc("When set to COALESCE, if two bucketed tables with the different number of buckets " +
Review comment:
Thanks for the suggestion. I updated.
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683158107
**[Test build #128003 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128003/testReport)** for PR 29473 at commit [`5665bc1`](https://github.com/apache/spark/commit/5665bc1107d6f9f06d1663a6f1ba8fa2ef5491e5).
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-675797802
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683385028
----------------------------------------------------------------
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.
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
[GitHub] [spark] maropu commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
maropu commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-682271192
btw, still `Draft`?
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683248506
Merged build finished. Test FAILed.
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-675876095
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683385078
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683241374
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683211853
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683385093
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/128027/
Test FAILed.
----------------------------------------------------------------
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.
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
[GitHub] [spark] c21 commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-678705963
@imback82 - thanks for working on this.
Seeing this is still marked as draft. Please change it once it's ready for review. 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.
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
[GitHub] [spark] SparkQA removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683375706
----------------------------------------------------------------
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.
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
[GitHub] [spark] maropu commented on a change in pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #29473:
URL: https://github.com/apache/spark/pull/29473#discussion_r478778853
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/bucketing/CoalesceOrRepartitionBucketsInJoin.scala
##########
@@ -83,23 +82,39 @@ case class CoalesceBucketsInJoin(conf: SQLConf) extends Rule[SparkPlan] {
}
def apply(plan: SparkPlan): SparkPlan = {
- if (!conf.coalesceBucketsInJoinEnabled) {
+ if (!conf.coalesceBucketsInJoinEnabled && !conf.repartitionBucketsInJoinEnabled) {
return plan
}
+ if (conf.coalesceBucketsInJoinEnabled && conf.repartitionBucketsInJoinEnabled) {
+ throw new AnalysisException("Both 'spark.sql.bucketing.coalesceBucketsInJoin.enabled' and " +
+ "'spark.sql.bucketing.repartitionBucketsInJoin.enabled' cannot be set to true at the" +
+ "same time")
Review comment:
For example;
```
object BucketReadStrategyMode extends Enumeration {
val COALESCING, REPARTITIONING, AUTOMATIC, OFF = Value
}
val BUCKET_READ_STORATEGY_MODE =
.buildConf("...")
.version("3.1.0")
.stringConf
.transform(_.toUpperCase(Locale.ROOT))
.checkValues(BucketReadStrategyMode.values.map(_.toString))
.createWithDefault(BucketReadStrategyMode.OFF.toString)
```
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683219778
----------------------------------------------------------------
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.
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
[GitHub] [spark] imback82 commented on a change in pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
imback82 commented on a change in pull request #29473:
URL: https://github.com/apache/spark/pull/29473#discussion_r478714367
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -166,15 +167,24 @@ case class FileSourceScanExec(
requiredSchema: StructType,
partitionFilters: Seq[Expression],
optionalBucketSet: Option[BitSet],
- optionalNumCoalescedBuckets: Option[Int],
+ optionalNewNumBuckets: Option[Int],
dataFilters: Seq[Expression],
tableIdentifier: Option[TableIdentifier])
extends DataSourceScanExec {
// Note that some vals referring the file-based relation are lazy intentionally
// so that this plan can be canonicalized on executor side too. See SPARK-23731.
override lazy val supportsColumnar: Boolean = {
- relation.fileFormat.supportBatch(relation.sparkSession, schema)
+ // `RepartitioningBucketRDD` converts columnar batches to rows to calculate bucket id for each
+ // row, thus columnar is not supported when `RepartitioningBucketRDD` is used to avoid
+ // conversions from batches to rows and back to batches.
+ relation.fileFormat.supportBatch(relation.sparkSession, schema) && !isRepartitioningBuckets
+ }
+
+ @transient private lazy val isRepartitioningBuckets: Boolean = {
Review comment:
I followed the same style from `override lazy val supportsColumnar: Boolean`, etc. Is this still not needed?
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683158565
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683375280
**[Test build #128027 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128027/testReport)** for PR 29473 at commit [`2c4925b`](https://github.com/apache/spark/commit/2c4925ba5258b3728608e77a3fb56250c1c15bbb).
----------------------------------------------------------------
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.
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
[GitHub] [spark] imback82 commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
imback82 commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-682212843
> Just a question; we cannot set a simpler rule to select which strategy (reapportioning or coalescing) we use when reading buckets? I think it is a bit annoying to set true to `repartitionBucketsInJoin.enabled` or `coalesceBucketsInJoin.enabled` data-by-data.
Good point. One use case for repartition over coalesce is when there are enough cores available in the cluster, not to reduce the parallelism by coalescing. @c21, did you observe any patterns or heuristics on your workloads where repartition is preferred?
> For example, if one has too small buckets (this case is extreme though), repartitioning for higher cardinality might not be able to help for better performance.
This is still guarded by `spark.sql.bucketing.coalesceOrRepartitionBucketsInJoin.maxBucketRatio`, so this scenario is a little less concerning?
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683227452
----------------------------------------------------------------
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.
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
[GitHub] [spark] github-actions[bot] commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-741340243
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683385092
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683375765
----------------------------------------------------------------
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.
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
[GitHub] [spark] imback82 commented on a change in pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
imback82 commented on a change in pull request #29473:
URL: https://github.com/apache/spark/pull/29473#discussion_r479590003
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/bucketing/CoalesceOrRepartitionBucketsInJoin.scala
##########
@@ -83,23 +82,39 @@ case class CoalesceBucketsInJoin(conf: SQLConf) extends Rule[SparkPlan] {
}
def apply(plan: SparkPlan): SparkPlan = {
- if (!conf.coalesceBucketsInJoinEnabled) {
+ if (!conf.coalesceBucketsInJoinEnabled && !conf.repartitionBucketsInJoinEnabled) {
return plan
}
+ if (conf.coalesceBucketsInJoinEnabled && conf.repartitionBucketsInJoinEnabled) {
+ throw new AnalysisException("Both 'spark.sql.bucketing.coalesceBucketsInJoin.enabled' and " +
+ "'spark.sql.bucketing.repartitionBucketsInJoin.enabled' cannot be set to true at the" +
+ "same time")
Review comment:
Thanks for the suggestion! I think the new config makes more sense. I renamed few, and let me know if it doesn't make sense.
Btw, do you think I can introduce `AUTOMATIC` as a follow up since this PR is sizable? Let me know if you want to see it in this PR. 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.
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
[GitHub] [spark] maropu commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
maropu commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-682271915
Also, I think its better to describe performance numbers for this proposed idea in the PR description above.
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683211853
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683375370
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683227251
**[Test build #128012 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128012/testReport)** for PR 29473 at commit [`7481e36`](https://github.com/apache/spark/commit/7481e36d8781e869a0dc558e0af5d358a56ab150).
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683219393
**[Test build #128003 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128003/testReport)** for PR 29473 at commit [`5665bc1`](https://github.com/apache/spark/commit/5665bc1107d6f9f06d1663a6f1ba8fa2ef5491e5).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683158107
**[Test build #128003 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128003/testReport)** for PR 29473 at commit [`5665bc1`](https://github.com/apache/spark/commit/5665bc1107d6f9f06d1663a6f1ba8fa2ef5491e5).
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683248506
----------------------------------------------------------------
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.
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
[GitHub] [spark] c21 commented on a change in pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #29473:
URL: https://github.com/apache/spark/pull/29473#discussion_r479612474
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/bucketing/CoalesceOrRepartitionBucketsInJoin.scala
##########
@@ -27,45 +27,48 @@ import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.{FileSourceScanExec, FilterExec, ProjectExec, SparkPlan}
import org.apache.spark.sql.execution.joins.{BaseJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.SQLConf.BucketReadStrategyInJoin
/**
- * This rule coalesces one side of the `SortMergeJoin` and `ShuffledHashJoin`
+ * This rule coalesces or repartitions one side of the `SortMergeJoin` and `ShuffledHashJoin`
* if the following conditions are met:
* - Two bucketed tables are joined.
* - Join keys match with output partition expressions on their respective sides.
* - The larger bucket number is divisible by the smaller bucket number.
- * - COALESCE_BUCKETS_IN_JOIN_ENABLED is set to true.
* - The ratio of the number of buckets is less than the value set in
- * COALESCE_BUCKETS_IN_JOIN_MAX_BUCKET_RATIO.
+ * COALESCE_OR_REPARTITION_BUCKETS_IN_JOIN_MAX_BUCKET_RATIO.
Review comment:
nit: shouldn't it be `BUCKET_READ_STRATEGY_IN_JOIN_MAX_BUCKET_RATIO` ?
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -2655,24 +2655,34 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
- val COALESCE_BUCKETS_IN_JOIN_ENABLED =
- buildConf("spark.sql.bucketing.coalesceBucketsInJoin.enabled")
- .doc("When true, if two bucketed tables with the different number of buckets are joined, " +
- "the side with a bigger number of buckets will be coalesced to have the same number " +
- "of buckets as the other side. Bigger number of buckets is divisible by the smaller " +
- "number of buckets. Bucket coalescing is applied to sort-merge joins and " +
- "shuffled hash join. Note: Coalescing bucketed table can avoid unnecessary shuffling " +
- "in join, but it also reduces parallelism and could possibly cause OOM for " +
- "shuffled hash join.")
- .version("3.1.0")
- .booleanConf
- .createWithDefault(false)
+ object BucketReadStrategyInJoin extends Enumeration {
+ val COALESCE, REPARTITION, OFF = Value
+ }
- val COALESCE_BUCKETS_IN_JOIN_MAX_BUCKET_RATIO =
- buildConf("spark.sql.bucketing.coalesceBucketsInJoin.maxBucketRatio")
- .doc("The ratio of the number of two buckets being coalesced should be less than or " +
- "equal to this value for bucket coalescing to be applied. This configuration only " +
- s"has an effect when '${COALESCE_BUCKETS_IN_JOIN_ENABLED.key}' is set to true.")
+ val BUCKET_READ_STRATEGY_IN_JOIN =
+ buildConf("spark.sql.bucketing.bucketReadStrategyInJoin")
+ .doc("When set to COALESCE, if two bucketed tables with the different number of buckets " +
Review comment:
nit: shall we first mention the allowed values to be "one of COALESCE, REPARTITION, OFF"? User might not follow exactly after long description here. Also probably worth to mention by default is "OFF" where we do not coalesce and repartition.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -548,22 +560,42 @@ case class FileSourceScanExec(
filesGroupedToBuckets
}
- val filePartitions = optionalNumCoalescedBuckets.map { numCoalescedBuckets =>
- logInfo(s"Coalescing to ${numCoalescedBuckets} buckets")
- val coalescedBuckets = prunedFilesGroupedToBuckets.groupBy(_._1 % numCoalescedBuckets)
- Seq.tabulate(numCoalescedBuckets) { bucketId =>
- val partitionedFiles = coalescedBuckets.get(bucketId).map {
- _.values.flatten.toArray
- }.getOrElse(Array.empty)
- FilePartition(bucketId, partitionedFiles)
- }
- }.getOrElse {
- Seq.tabulate(bucketSpec.numBuckets) { bucketId =>
+ if (optionalNewNumBuckets.isEmpty) {
+ val filePartitions = Seq.tabulate(bucketSpec.numBuckets) { bucketId =>
FilePartition(bucketId, prunedFilesGroupedToBuckets.getOrElse(bucketId, Array.empty))
}
+ new FileScanRDD(fsRelation.sparkSession, readFile, filePartitions)
+ } else {
+ val newNumBuckets = optionalNewNumBuckets.get
+ if (newNumBuckets < bucketSpec.numBuckets) {
+ assert(bucketSpec.numBuckets % newNumBuckets == 0)
+ logInfo(s"Coalescing to $newNumBuckets buckets from ${bucketSpec.numBuckets} buckets")
+ val coalescedBuckets = prunedFilesGroupedToBuckets.groupBy(_._1 % newNumBuckets)
+ val filePartitions = Seq.tabulate(newNumBuckets) { bucketId =>
+ val partitionedFiles = coalescedBuckets
+ .get(bucketId)
+ .map(_.values.flatten.toArray)
+ .getOrElse(Array.empty)
+ FilePartition(bucketId, partitionedFiles)
+ }
+ new FileScanRDD(fsRelation.sparkSession, readFile, filePartitions)
+ } else {
+ assert(newNumBuckets % bucketSpec.numBuckets == 0)
+ logInfo(s"Repartitioning to $newNumBuckets buckets from ${bucketSpec.numBuckets} buckets")
+ val filePartitions = Seq.tabulate(newNumBuckets) { bucketId =>
+ FilePartition(
+ bucketId,
+ prunedFilesGroupedToBuckets.getOrElse(bucketId % bucketSpec.numBuckets, Array.empty))
+ }
+ // There are now more files to be read.
+ val filesNum = filePartitions.map(_.files.size.toLong).sum
+ val filesSize = filePartitions.map(_.files.map(_.length).sum).sum
+ driverMetrics("numFiles") = filesNum
Review comment:
per `setFilesNumAndSizeMetric`, should we set `staticFilesNum` here or `numFiles ` ?
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -2655,24 +2655,34 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
- val COALESCE_BUCKETS_IN_JOIN_ENABLED =
- buildConf("spark.sql.bucketing.coalesceBucketsInJoin.enabled")
- .doc("When true, if two bucketed tables with the different number of buckets are joined, " +
- "the side with a bigger number of buckets will be coalesced to have the same number " +
- "of buckets as the other side. Bigger number of buckets is divisible by the smaller " +
- "number of buckets. Bucket coalescing is applied to sort-merge joins and " +
- "shuffled hash join. Note: Coalescing bucketed table can avoid unnecessary shuffling " +
- "in join, but it also reduces parallelism and could possibly cause OOM for " +
- "shuffled hash join.")
- .version("3.1.0")
- .booleanConf
- .createWithDefault(false)
+ object BucketReadStrategyInJoin extends Enumeration {
+ val COALESCE, REPARTITION, OFF = Value
+ }
- val COALESCE_BUCKETS_IN_JOIN_MAX_BUCKET_RATIO =
- buildConf("spark.sql.bucketing.coalesceBucketsInJoin.maxBucketRatio")
- .doc("The ratio of the number of two buckets being coalesced should be less than or " +
- "equal to this value for bucket coalescing to be applied. This configuration only " +
- s"has an effect when '${COALESCE_BUCKETS_IN_JOIN_ENABLED.key}' is set to true.")
+ val BUCKET_READ_STRATEGY_IN_JOIN =
+ buildConf("spark.sql.bucketing.bucketReadStrategyInJoin")
Review comment:
nit: shall we have a name also consistent with existing config "spark.sql.sources.bucketing", e.g. "spark.sql.sources.bucketing.readStrategyInJoin". No big deal, but "bucketing.bucket..." seems a little bit verbose. Point out here because users might depend on this config for bucketing optimization and raise questions for developers with this config.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -164,15 +165,24 @@ case class FileSourceScanExec(
requiredSchema: StructType,
partitionFilters: Seq[Expression],
optionalBucketSet: Option[BitSet],
- optionalNumCoalescedBuckets: Option[Int],
+ optionalNewNumBuckets: Option[Int],
dataFilters: Seq[Expression],
tableIdentifier: Option[TableIdentifier])
extends DataSourceScanExec {
// Note that some vals referring the file-based relation are lazy intentionally
// so that this plan can be canonicalized on executor side too. See SPARK-23731.
override lazy val supportsColumnar: Boolean = {
- relation.fileFormat.supportBatch(relation.sparkSession, schema)
+ // `RepartitioningBucketRDD` converts columnar batches to rows to calculate bucket id for each
+ // row, thus columnar is not supported when `RepartitioningBucketRDD` is used to avoid
+ // conversions from batches to rows and back to batches.
+ relation.fileFormat.supportBatch(relation.sparkSession, schema) && !isRepartitioningBuckets
+ }
+
+ @transient private lazy val isRepartitioningBuckets: Boolean = {
+ relation.bucketSpec.isDefined &&
+ optionalNewNumBuckets.isDefined &&
+ optionalNewNumBuckets.get > relation.bucketSpec.get.numBuckets
Review comment:
we don't want to do the repartition if user disables bucketing right?
better to do with `bucketedScan && optionalNewNumBuckets.isDefined && optionalNewNumBuckets.get > relation.bucketSpec.get.numBuckets` ?
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -2655,24 +2655,34 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
- val COALESCE_BUCKETS_IN_JOIN_ENABLED =
- buildConf("spark.sql.bucketing.coalesceBucketsInJoin.enabled")
- .doc("When true, if two bucketed tables with the different number of buckets are joined, " +
- "the side with a bigger number of buckets will be coalesced to have the same number " +
- "of buckets as the other side. Bigger number of buckets is divisible by the smaller " +
- "number of buckets. Bucket coalescing is applied to sort-merge joins and " +
- "shuffled hash join. Note: Coalescing bucketed table can avoid unnecessary shuffling " +
- "in join, but it also reduces parallelism and could possibly cause OOM for " +
- "shuffled hash join.")
- .version("3.1.0")
- .booleanConf
- .createWithDefault(false)
+ object BucketReadStrategyInJoin extends Enumeration {
+ val COALESCE, REPARTITION, OFF = Value
+ }
- val COALESCE_BUCKETS_IN_JOIN_MAX_BUCKET_RATIO =
- buildConf("spark.sql.bucketing.coalesceBucketsInJoin.maxBucketRatio")
- .doc("The ratio of the number of two buckets being coalesced should be less than or " +
- "equal to this value for bucket coalescing to be applied. This configuration only " +
- s"has an effect when '${COALESCE_BUCKETS_IN_JOIN_ENABLED.key}' is set to true.")
+ val BUCKET_READ_STRATEGY_IN_JOIN =
+ buildConf("spark.sql.bucketing.bucketReadStrategyInJoin")
+ .doc("When set to COALESCE, if two bucketed tables with the different number of buckets " +
+ "are joined, the side with a bigger number of buckets will be coalesced to have the same " +
+ "number of buckets as the other side. When set to REPARTITION, the side with a bigger " +
+ "number of buckets will be repartitioned to have the same number of buckets as the other " +
+ "side. The bigger number of buckets must be divisible by the smaller number of buckets, " +
+ "and the strategy is applied to sort-merge joins and shuffled hash joins. " +
+ "Note: Coalescing bucketed table can avoid unnecessary shuffle in join, but it also " +
+ "reduces parallelism and could possibly cause OOM for shuffled hash join. Repartitioning " +
+ "bucketed table avoids unnecessary shuffle in join while maintaining the parallelism " +
+ "at the cost of reading duplicate data.")
+ .version("3.1.0")
+ .stringConf
+ .transform(_.toUpperCase(Locale.ROOT))
+ .checkValues(BucketReadStrategyInJoin.values.map(_.toString))
+ .createWithDefault(BucketReadStrategyInJoin.OFF.toString)
+
+ val BUCKET_READ_STRATEGY_IN_JOIN_MAX_BUCKET_RATIO =
+ buildConf("spark.sql.bucketing.bucketReadStrategyInJoin.maxBucketRatio")
Review comment:
nit: same as above, might be just ""spark.sql.sources.bucketing.readStrategyInJoinMaxBucketRatio" ?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -314,7 +324,7 @@ case class FileSourceScanExec(
val singleFilePartitions = bucketToFilesGrouping.forall(p => p._2.length <= 1)
// TODO SPARK-24528 Sort order is currently ignored if buckets are coalesced.
- if (singleFilePartitions && optionalNumCoalescedBuckets.isEmpty) {
+ if (singleFilePartitions && (optionalNewNumBuckets.isEmpty || isRepartitioningBuckets)) {
Review comment:
we don't need `|| isRepartitioningBuckets` right?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
##########
@@ -548,22 +560,42 @@ case class FileSourceScanExec(
filesGroupedToBuckets
}
- val filePartitions = optionalNumCoalescedBuckets.map { numCoalescedBuckets =>
- logInfo(s"Coalescing to ${numCoalescedBuckets} buckets")
- val coalescedBuckets = prunedFilesGroupedToBuckets.groupBy(_._1 % numCoalescedBuckets)
- Seq.tabulate(numCoalescedBuckets) { bucketId =>
- val partitionedFiles = coalescedBuckets.get(bucketId).map {
- _.values.flatten.toArray
- }.getOrElse(Array.empty)
- FilePartition(bucketId, partitionedFiles)
- }
- }.getOrElse {
- Seq.tabulate(bucketSpec.numBuckets) { bucketId =>
+ if (optionalNewNumBuckets.isEmpty) {
+ val filePartitions = Seq.tabulate(bucketSpec.numBuckets) { bucketId =>
FilePartition(bucketId, prunedFilesGroupedToBuckets.getOrElse(bucketId, Array.empty))
}
+ new FileScanRDD(fsRelation.sparkSession, readFile, filePartitions)
+ } else {
+ val newNumBuckets = optionalNewNumBuckets.get
+ if (newNumBuckets < bucketSpec.numBuckets) {
+ assert(bucketSpec.numBuckets % newNumBuckets == 0)
+ logInfo(s"Coalescing to $newNumBuckets buckets from ${bucketSpec.numBuckets} buckets")
+ val coalescedBuckets = prunedFilesGroupedToBuckets.groupBy(_._1 % newNumBuckets)
+ val filePartitions = Seq.tabulate(newNumBuckets) { bucketId =>
+ val partitionedFiles = coalescedBuckets
+ .get(bucketId)
+ .map(_.values.flatten.toArray)
+ .getOrElse(Array.empty)
+ FilePartition(bucketId, partitionedFiles)
+ }
+ new FileScanRDD(fsRelation.sparkSession, readFile, filePartitions)
+ } else {
+ assert(newNumBuckets % bucketSpec.numBuckets == 0)
+ logInfo(s"Repartitioning to $newNumBuckets buckets from ${bucketSpec.numBuckets} buckets")
+ val filePartitions = Seq.tabulate(newNumBuckets) { bucketId =>
+ FilePartition(
+ bucketId,
+ prunedFilesGroupedToBuckets.getOrElse(bucketId % bucketSpec.numBuckets, Array.empty))
+ }
+ // There are now more files to be read.
+ val filesNum = filePartitions.map(_.files.size.toLong).sum
+ val filesSize = filePartitions.map(_.files.map(_.length).sum).sum
+ driverMetrics("numFiles") = filesNum
+ driverMetrics("filesSize") = filesSize
+ new BucketRepartitioningRDD(
+ fsRelation.sparkSession, readFile, filePartitions, bucketSpec, newNumBuckets, output)
Review comment:
nit: we can pass `outputPartitioning.asInstanceOf[HashPartitioning].partitionIdExpression`, so do not need to re-do logic to get `bucketIdExpression` in `BucketRepartitioningRDD`.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/bucketing/BucketRepartitioningRDD.scala
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.bucketing
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.{Partition, TaskContext}
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.catalog.BucketSpec
+import org.apache.spark.sql.catalyst.expressions.{Attribute, UnsafeProjection}
+import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.execution.datasources.{FilePartition, FileScanRDD, PartitionedFile}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+
+/**
+ * An RDD that filters out the rows that do not belong to the current bucket file being read.
+ */
+private[spark] class BucketRepartitioningRDD(
+ @transient private val sparkSession: SparkSession,
+ readFunction: PartitionedFile => Iterator[InternalRow],
+ @transient override val filePartitions: Seq[FilePartition],
+ bucketSpec: BucketSpec,
+ numRepartitionedBuckets: Int,
+ output: Seq[Attribute])
+ extends FileScanRDD(sparkSession, readFunction, filePartitions) {
+ assert(numRepartitionedBuckets > bucketSpec.numBuckets)
Review comment:
nit: probably just one assert for `numRepartitionedBuckets > bucketSpec.numBuckets && numRepartitionedBuckets % bucketSpec.numBuckets == 0`.
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683158565
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA commented on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683248360
**[Test build #128012 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128012/testReport)** for PR 29473 at commit [`7481e36`](https://github.com/apache/spark/commit/7481e36d8781e869a0dc558e0af5d358a56ab150).
* This patch **fails due to an unknown error code, -9**.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
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.
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
[GitHub] [spark] SparkQA removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-675797404
**[Test build #127609 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127609/testReport)** for PR 29473 at commit [`21882ab`](https://github.com/apache/spark/commit/21882abe495ce6c1fff297ec7d52928f13a2fcc9).
----------------------------------------------------------------
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.
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
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29473: [SPARK-32656][SQL] Repartition bucketed tables for sort merge join / shuffled hash join if applicable
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29473:
URL: https://github.com/apache/spark/pull/29473#issuecomment-683248507
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/128012/
Test FAILed.
----------------------------------------------------------------
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.
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