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
Prev Previous commit
Address comments.
  • Loading branch information
viirya committed Sep 8, 2017
commit ecdfb7db34d0d01e357bff0d32b62137ef0ae735
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ abstract class Optimizer(sessionCatalog: SessionCatalog)

// Check for structural integrity of the plan in test mode. Currently we only check if a plan is
// still resolved after the execution of each rule.
override protected def planChecker(plan: LogicalPlan): Boolean = {
override protected def isPlanIntegral(plan: LogicalPlan): Boolean = {
Utils.isTesting && plan.resolved
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -64,12 +64,12 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
protected def batches: Seq[Batch]

/**
* Defines a check function which checks for structural integrity of the plan after the execution
* Defines a check function that checks for structural integrity of the plan after the execution
* of each rule. For example, we can check whether a plan is still resolved after each rule in
* `Optimizer`, so we can catch rules that return invalid plans. The check function will returns
* `Optimizer`, so we can catch rules that return invalid plans. The check function returns
* `false` if the given plan doesn't pass the structural integrity check.
*/
protected def planChecker(plan: TreeType): Boolean = true
protected def isPlanIntegral(plan: TreeType): Boolean = true

/**
* Executes the batches of rules defined by the subclass. The batches are executed serially
Expand Down Expand Up @@ -102,7 +102,7 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
}

// Run the structural integrity checker against the plan after each rule.
if (!planChecker(result)) {
if (!isPlanIntegral(result)) {
val message = s"After applying rule ${rule.ruleName} in batch ${batch.name}, " +
"the structural integrity of the plan is broken."
throw new TreeNodeException(result, message, null)
Copy link
Member

Choose a reason for hiding this comment

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

move the exception throwing logics into the planChecker ?

Copy link
Member

Choose a reason for hiding this comment

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

nvm. The message also has rule and batch names.

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.rules._
import org.apache.spark.sql.internal.SQLConf


class OptimizerSICheckerkSuite extends PlanTest {
class OptimizerStructuralIntegrityCheckerSuite extends PlanTest {

object OptimizeRuleBreakSI extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ class RuleExecutorSuite extends SparkFunSuite {

test("structural integrity checker") {
object WithSIChecker extends RuleExecutor[Expression] {
override protected def planChecker(expr: Expression): Boolean = expr match {
override protected def isPlanIntegral(expr: Expression): Boolean = expr match {
case IntegerLiteral(_) => true
case _ => false
}
Expand Down