Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-43021][SQL] CoalesceBucketsInJoin not work when using AQE #40688

Closed
wants to merge 4 commits into from
Closed

[SPARK-43021][SQL] CoalesceBucketsInJoin not work when using AQE #40688

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5ae6809
SPARK-43021
zzzzming95 Apr 10, 2023
527e6d8
update AdaptiveSparkPlanExec.scala
zzzzming95 Apr 12, 2023
36257ae
BucketedReadSuite.scala
zzzzming95 Apr 12, 2023
2a55aac
update BucketedReadSuite.scala
zzzzming95 Apr 13, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ import org.apache.spark.sql.catalyst.util.sideBySide
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec._
import org.apache.spark.sql.execution.bucketing.DisableUnnecessaryBucketedScan
import org.apache.spark.sql.execution.bucketing.{CoalesceBucketsInJoin, DisableUnnecessaryBucketedScan}
import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
import org.apache.spark.sql.execution.exchange._
import org.apache.spark.sql.execution.ui.{SparkListenerSQLAdaptiveExecutionUpdate, SparkListenerSQLAdaptiveSQLMetricUpdates, SQLPlanMetric}
Expand Down Expand Up @@ -118,6 +118,7 @@ case class AdaptiveSparkPlanExec(
val ensureRequirements =
EnsureRequirements(requiredDistribution.isDefined, requiredDistribution)
Seq(
zzzzming95 marked this conversation as resolved.
Show resolved Hide resolved
CoalesceBucketsInJoin,
Copy link
Contributor

Choose a reason for hiding this comment

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

shall we put it in queryStageOptimizerRules?

Copy link
Contributor

Choose a reason for hiding this comment

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

rules in queryStageOptimizerRules are invoked less often which is more efficient. The rule CoalesceBucketsInJoin does not change plan partitioning and seems can be put in queryStageOptimizerRules

Copy link
Contributor Author

Choose a reason for hiding this comment

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

In my test , the UT run failed if CoalesceBucketsInJoin add in queryStageOptimizerRules .

Copy link
Contributor

@cloud-fan cloud-fan Apr 10, 2023 •
edited
Loading

Choose a reason for hiding this comment

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

Can we spend a bit of time understanding why? Then we can write a code comment to explain it and future developers won't try to move this rule to queryStageOptimizerRules ever.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Yeah , I will provide detailed information and supplement it .

Copy link
Contributor Author

@zzzzming95 zzzzming95 Apr 11, 2023 •
edited
Loading

Choose a reason for hiding this comment

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

Because queryStageOptimizerRules is not applied at the beginning of the init plan. Instead, they are applied in the createQueryStages() method. And createQueryStages() is bottom-up, which causes the exchange to be eliminated to be wrapped in a layer of ShuffleQueryStage first, making CoalesceBucketsInJoin unrecognizable. And I have added these to the notes at the top. thanks @cloud-fan

Copy link
Member

Choose a reason for hiding this comment

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

CoalesceBucketsInJoin should before EnsureRequirements.

RemoveRedundantProjects,
ensureRequirements,
AdjustShuffleExchangePosition,
Expand Down
18 changes: 17 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
import org.apache.spark.sql.execution.{FileSourceScanExec, SortExec, SparkPlan}
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AdaptiveSparkPlanHelper, DisableAdaptiveExecution}
import org.apache.spark.sql.execution.datasources.BucketingUtils
import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
import org.apache.spark.sql.execution.exchange.{Exchange, ShuffleExchangeExec}
import org.apache.spark.sql.execution.joins.SortMergeJoinExec
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
Expand Down Expand Up @@ -1050,4 +1050,20 @@ abstract class BucketedReadSuite extends QueryTest with SQLTestUtils with Adapti
}
}
}

test("SPARK-43021: Make coalesceBucketsInJoin effective enable AQE") {
zzzzming95 marked this conversation as resolved.
Show resolved Hide resolved
withSQLConf(
SQLConf.COALESCE_BUCKETS_IN_JOIN_ENABLED.key -> "true",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
withTable("bucketed_table1", "bucketed_table2") {
df1.write.format("parquet").bucketBy(4, "i").saveAsTable("bucketed_table1")
df2.write.format("parquet").bucketBy(2, "i").saveAsTable("bucketed_table2")
val t1 = spark.table("bucketed_table1")
val t2 = spark.table("bucketed_table2")
val plan = t1.join(t2, t1("i") === t2("i")).queryExecution.executedPlan
assert(collect(plan) { case _: Exchange => true }.size === 0)
}
}
}
}