Address comment.

Author: viirya

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -29,6 +29,7 @@ import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules._
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
+import org.apache.spark.util.Utils
 
 /**
  * Abstract class all optimizers should inherit of, contains the standard batches (extending
@@ -39,9 +40,9 @@ 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: Option[LogicalPlan => Boolean] = Some(
-    (plan: LogicalPlan) => plan.resolved
-  )
+  override protected def planChecker(plan: LogicalPlan): Boolean = {
+    Utils.isTesting && plan.resolved
+  }
 
   protected def fixedPoint = FixedPoint(SQLConf.get.optimizerMaxIterations)
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala

@@ -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 in test mode 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
-   * returns `false` if the given plan doesn't pass the structural integrity check.
+   * Defines a check function which 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
+   * `false` if the given plan doesn't pass the structural integrity check.
    */
-  protected def planChecker: Option[TreeType => Boolean] = None
+  protected def planChecker(plan: TreeType): Boolean = true
 
   /**
    * Executes the batches of rules defined by the subclass. The batches are executed serially
@@ -101,8 +101,8 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
                 """.stripMargin)
             }
 
-            // In test mode, run the structural integrity checker against the plan after each rule.
-            if (Utils.isTesting && !planChecker.map(_.apply(result)).getOrElse(true)) {
+            // Run the structural integrity checker against the plan after each rule.
+            if (!planChecker(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)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/trees/RuleExecutorSuite.scala

@@ -59,14 +59,10 @@ class RuleExecutorSuite extends SparkFunSuite {
 
   test("structural integrity checker") {
     object WithSIChecker extends RuleExecutor[Expression] {
-      override protected def planChecker: Option[Expression => Boolean] = Some(
-        (expr: Expression) => {
-          expr match {
-            case IntegerLiteral(_) => true
-            case _ => false
-          }
-        }
-      )
+      override protected def planChecker(expr: Expression): Boolean = expr match {
+        case IntegerLiteral(_) => true
+        case _ => false
+      }
       val batches = Batch("once", Once, DecrementLiterals) :: Nil
     }