Skip to content

[SPARK-51812][SQL] Remove redundant parameters of some methods in QueryExecution #50598

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
panbingkun wants to merge 2 commits into from
Closed

[SPARK-51812][SQL] Remove redundant parameters of some methods in QueryExecution #50598

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6705fc5
[SPARK-51812][SQL] Remove redundant parameters of method `QueryExecut…
panbingkun Apr 16, 2025
c0b9f18
update
panbingkun Apr 16, 2025
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
7 changes: 3 additions & 4 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,7 +209,7 @@ class QueryExecution(
executePhase(QueryPlanningTracker.PLANNING) {
// Clone the logical plan here, in case the planner rules change the states of the logical
// plan.
QueryExecution.createSparkPlan(sparkSession, planner, optimizedPlan.clone())
QueryExecution.createSparkPlan(planner, optimizedPlan.clone())
}
}

Expand Down Expand Up @@ -574,7 +574,6 @@ object QueryExecution {
* Note that the returned physical plan still needs to be prepared for execution.
*/
def createSparkPlan(
sparkSession: SparkSession,
planner: SparkPlanner,
plan: LogicalPlan): SparkPlan = {
// TODO: We use next(), i.e. take the first plan returned by the planner, here for now,
Expand All @@ -594,7 +593,7 @@ object QueryExecution {
* [[SparkPlan]] for execution.
*/
def prepareExecutedPlan(spark: SparkSession, plan: LogicalPlan): SparkPlan = {
val sparkPlan = createSparkPlan(spark, spark.sessionState.planner, plan.clone())
val sparkPlan = createSparkPlan(spark.sessionState.planner, plan.clone())
prepareExecutedPlan(spark, sparkPlan)
}

Expand All @@ -606,7 +605,7 @@ object QueryExecution {
session: SparkSession,

@summaryzb summaryzb Apr 16, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

This session can be removed too, since it's the same SparkSession of the AdaptiveExecutionContext

@panbingkun panbingkun Apr 16, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Yeah, that's right.

plan: LogicalPlan,
context: AdaptiveExecutionContext): SparkPlan = {
val sparkPlan = createSparkPlan(session, session.sessionState.planner, plan.clone())
val sparkPlan = createSparkPlan(session.sessionState.planner, plan.clone())
val preparationRules = preparations(session, Option(InsertAdaptiveSparkPlan(context)), true)
prepareForExecution(preparationRules, sparkPlan.clone())
}
Expand Down
2 changes: 1 addition & 1 deletion ...core/src/main/scala/org/apache/spark/sql/execution/adaptive/InsertAdaptiveSparkPlan.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,7 +153,7 @@ case class InsertAdaptiveSparkPlan(
// Apply the same instance of this rule to sub-queries so that sub-queries all share the
// same `stageCache` for Exchange reuse.
this.applyInternal(
QueryExecution.createSparkPlan(adaptiveExecutionContext.session,
QueryExecution.createSparkPlan(
adaptiveExecutionContext.session.sessionState.planner, plan.clone()), true)
}

Expand Down
3 changes: 1 addition & 2 deletions .../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 @@ -55,8 +55,7 @@ case class PlanDynamicPruningFilters(sparkSession: SparkSession) extends Rule[Sp
plan.transformAllExpressionsWithPruning(_.containsPattern(DYNAMIC_PRUNING_SUBQUERY)) {
case DynamicPruningSubquery(
value, buildPlan, buildKeys, broadcastKeyIndices, onlyInBroadcast, exprId, _) =>
val sparkPlan = QueryExecution.createSparkPlan(
sparkSession, sparkSession.sessionState.planner, buildPlan)
val sparkPlan = QueryExecution.createSparkPlan(sparkSession.sessionState.planner, buildPlan)
// Using `sparkPlan` is a little hacky as it is based on the assumption that this rule is
// the first to be applied (apart from `InsertAdaptiveSparkPlan`).
val canReuseExchange = conf.exchangeReuseEnabled && buildKeys.nonEmpty &&
Expand Down