Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -226,6 +226,21 @@ class SparkSessionExtensions {
preCBORules += builder
}

private[this] val earlyScanPushDownRules = mutable.Buffer.empty[RuleBuilder]

private[sql] def buildEarlyScanPushDownRules(session: SparkSession): Seq[Rule[LogicalPlan]] = {
earlyScanPushDownRules.map(_.apply(session)).toSeq
}

/**
* Inject an optimizer `Rule` builder that rewrites logical plans into the [[SparkSession]].
* The injected rules will be executed once after the operator optimization batch and
* after any push down optimization rules.
*/
def injectEarlyScanPushDownRules(builder: RuleBuilder): Unit = {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

nit: injectEarlyScanPushDownRules -> injectEarlyScanPushDownRule

earlyScanPushDownRules += builder
}

private[this] val plannerStrategyBuilders = mutable.Buffer.empty[StrategyBuilder]

private[sql] def buildPlannerStrategies(session: SparkSession): Seq[Strategy] = {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -270,7 +270,9 @@ abstract class BaseSessionStateBuilder(
*
* Note that this may NOT depend on the `optimizer` function.
*/
protected def customEarlyScanPushDownRules: Seq[Rule[LogicalPlan]] = Nil
protected def customEarlyScanPushDownRules: Seq[Rule[LogicalPlan]] = {
extensions.buildEarlyScanPushDownRules(session)
}

/**
* Custom rules for rewriting plans after operator optimization and before CBO.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,12 @@ class SparkSessionExtensionSuite extends SparkFunSuite {
}
}

test("SPARK-37518: inject a early scan push down rule") {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

nit: I think we only require JIRA id for bug fixes and regressions?

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.

Just references

withSession(Seq(_.injectEarlyScanPushDownRules(MyRule))) { session =>
assert(session.sessionState.optimizer.earlyScanPushDownRules.contains(MyRule(session)))
}
}

test("inject spark planner strategy") {
withSession(Seq(_.injectPlannerStrategy(MySparkStrategy))) { session =>
assert(session.sessionState.planner.strategies.contains(MySparkStrategy(session)))
Expand Down