Skip to content

[SPARK-38959][SQL][FOLLOWUP] Do not optimize subqueries twice #38626

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

Closed
cloud-fan wants to merge 1 commit into from
Closed

[SPARK-38959][SQL][FOLLOWUP] Do not optimize subqueries twice #38626

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
6 changes: 4 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,10 @@ class SparkOptimizer(
Batch("Optimize Metadata Only Query", Once, OptimizeMetadataOnlyQuery(catalog)) :+
Batch("PartitionPruning", Once,
PartitionPruning,
RowLevelOperationRuntimeGroupFiltering,
OptimizeSubqueries) :+
// We can't run `OptimizeSubqueries` in this batch, as it will optimize the subqueries
// twice which may break some optimizer rules that can only be applied once. The rule below
// only invokes `OptimizeSubqueries` to optimize newly added subqueries.
Comment on lines +54 to +56
Copy link
Member

@viirya viirya Nov 12, 2022

Choose a reason for hiding this comment

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

Hm? This batch has only PartitionPruning and RowLevelOperationRuntimeGroupFiltering. What some optimizer rules are? PartitionPruning?

Copy link
Member

@viirya viirya Nov 12, 2022

Choose a reason for hiding this comment

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

Oh, you mean other Once batches in SparkOptimizer.defaultBatches?

Copy link
Member

@viirya viirya Nov 12, 2022

Choose a reason for hiding this comment

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

But in Optimizer where OptimizeSubqueries also runs, there are also other Once batches but seems fine?

Copy link
Contributor Author

@cloud-fan cloud-fan Nov 14, 2022

Choose a reason for hiding this comment

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

All the optimizer batches are optimizing the same query plan. If OptimizeSubqueries appears twice, it means the subqueries are optimized twice.

Note that, most optimizer rules don't optimize subqueries, they need OptimizeSubqueries to invoke the entire optimizer to optimize subqueries recursively.

Copy link
Contributor Author

@cloud-fan cloud-fan Nov 14, 2022

Choose a reason for hiding this comment

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

inspired by #38619 , maybe we don't need to invoke the entire optimizer, but just a few rules to optimize this subquery.

Copy link
Member

@viirya viirya Nov 14, 2022

Choose a reason for hiding this comment

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

All the optimizer batches are optimizing the same query plan. If OptimizeSubqueries appears twice, it means the subqueries are optimized twice.

Oh, got it, you actually mean OptimizeSubqueries is applied twice (here and Optimizer). I thought that by running OptimizeSubqueries itself here breaks some rules which cannot run twice.

new RowLevelOperationRuntimeGroupFiltering(OptimizeSubqueries)) :+
Batch("InjectRuntimeFilter", FixedPoint(1),
InjectRuntimeFilter) :+
Batch("MergeScalarSubqueries", Once,
Expand Down
2 changes: 1 addition & 1 deletion ...ain/scala/org/apache/spark/sql/execution/adaptive/PlanAdaptiveDynamicPruningFilters.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, HashedRelati
case class PlanAdaptiveDynamicPruningFilters(
rootPlan: AdaptiveSparkPlanExec) extends Rule[SparkPlan] with AdaptiveSparkPlanHelper {
def apply(plan: SparkPlan): SparkPlan = {
if (!conf.dynamicPartitionPruningEnabled && !conf.runtimeRowLevelOperationGroupFilterEnabled) {
if (!conf.dynamicPartitionPruningEnabled) {
return plan
}

Expand Down
2 changes: 1 addition & 1 deletion .../main/scala/org/apache/spark/sql/execution/dynamicpruning/PlanDynamicPruningFilters.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ case class PlanDynamicPruningFilters(sparkSession: SparkSession) extends Rule[Sp
}

override def apply(plan: SparkPlan): SparkPlan = {
if (!conf.dynamicPartitionPruningEnabled && !conf.runtimeRowLevelOperationGroupFilterEnabled) {
if (!conf.dynamicPartitionPruningEnabled) {
return plan
}

Expand Down
6 changes: 4 additions & 2 deletions ...rg/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,8 @@ import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Implicits, Dat
*
* Note this rule only applies to group-based row-level operations.
*/
object RowLevelOperationRuntimeGroupFiltering extends Rule[LogicalPlan] with PredicateHelper {
class RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPlan])
extends Rule[LogicalPlan] with PredicateHelper {

import DataSourceV2Implicits._

Expand All @@ -64,7 +65,8 @@ object RowLevelOperationRuntimeGroupFiltering extends Rule[LogicalPlan] with Pre
Filter(dynamicPruningCond, r)
}

replaceData.copy(query = newQuery)
// optimize subqueries to rewrite them as joins and trigger job planning
replaceData.copy(query = optimizeSubqueries(newQuery))
}

private def buildMatchingRowsPlan(
Expand Down