address comments.

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

@@ -21,7 +21,8 @@ import scala.collection.JavaConverters._
 
 import com.google.common.util.concurrent.AtomicLongMap
 
-import org.apache.spark.{Logging, SparkException}
+import org.apache.spark.Logging
+import org.apache.spark.sql.catalyst.errors.TreeNodeException
 import org.apache.spark.sql.catalyst.trees.TreeNode
 import org.apache.spark.sql.catalyst.util.sideBySide
 import org.apache.spark.util.Utils
@@ -51,18 +52,18 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
    */
   abstract class Strategy {
     def maxIterations: Int
-    def throws: Boolean
+    def throwsExceptionUponMaxIterations: Boolean
   }
 
   /** A strategy that only runs once. */
   case object Once extends Strategy {
-    val maxIterations = 1
-    val throws = false
+    override val maxIterations = 1
+    override val throwsExceptionUponMaxIterations = false
   }
 
   /** A strategy that runs until fix point or maxIterations times, whichever comes first. */
   case class FixedPoint(maxIterations: Int) extends Strategy {
-    override val throws: Boolean = if (Utils.isTesting) true else false
+    override val throwsExceptionUponMaxIterations: Boolean = if (Utils.isTesting) true else false
   }
   /** A batch of rules. */
   protected case class Batch(name: String, strategy: Strategy, rules: Rule[TreeType]*)
@@ -108,7 +109,11 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
           // Only log if this is a rule that is supposed to run more than once.
           if (iteration != 2) {
             val msg = s"Max iterations (${iteration - 1}) reached for batch ${batch.name}"
-            if (batch.strategy.throws) throw new SparkException(msg) else logTrace(msg)
+            if (batch.strategy.throwsExceptionUponMaxIterations) {
+              throw new TreeNodeException(curPlan, msg, null)
+            } else {
+              logTrace(msg)
+            }
           }
           continue = false
         }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala

@@ -33,7 +33,6 @@ import org.apache.spark.sql.types.StringType
 class ColumnPruningSuite extends PlanTest {
 
   object Optimize extends RuleExecutor[LogicalPlan] {
-    System.setProperty("spark.testing", "true")
     val batches = Batch("Column pruning", FixedPoint(100),
       ColumnPruning,
       CollapseProject) :: Nil

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/CombiningLimitsSuite.scala

@@ -26,7 +26,6 @@ import org.apache.spark.sql.catalyst.rules._
 class CombiningLimitsSuite extends PlanTest {
 
   object Optimize extends RuleExecutor[LogicalPlan] {
-    System.setProperty("spark.testing", "true")
     val batches =
       Batch("Filter Pushdown", FixedPoint(100),
         ColumnPruning) ::

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ComputeCurrentTimeSuite.scala

@@ -25,7 +25,6 @@ import org.apache.spark.sql.catalyst.rules.RuleExecutor
 import org.apache.spark.sql.catalyst.util.DateTimeUtils
 
 class ComputeCurrentTimeSuite extends PlanTest {
-  System.setProperty("spark.testing", "true")
   object Optimize extends RuleExecutor[LogicalPlan] {
     val batches = Seq(Batch("ComputeCurrentTime", Once, ComputeCurrentTime))
   }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/SimplifyConditionalSuite.scala

@@ -29,7 +29,6 @@ import org.apache.spark.sql.types.IntegerType
 class SimplifyConditionalSuite extends PlanTest with PredicateHelper {
 
   object Optimize extends RuleExecutor[LogicalPlan] {
-    System.setProperty("spark.testing", "true")
     val batches =
       Batch("SimplifyConditionals", FixedPoint(50), SimplifyConditionals) :: Nil
   }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/PlanTest.scala

@@ -26,6 +26,9 @@ import org.apache.spark.sql.catalyst.util._
  * Provides helper methods for comparing plans.
  */
 abstract class PlanTest extends SparkFunSuite with PredicateHelper {
+
+  System.setProperty("spark.testing", "true")
+
   /**
    * Since attribute references are given globally unique ids during analysis,
    * we must normalize them to check if two different queries are identical.

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

@@ -17,8 +17,10 @@
 
 package org.apache.spark.sql.catalyst.trees
 
-import org.apache.spark.{SparkException, SparkFunSuite}
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.errors.TreeNodeException
 import org.apache.spark.sql.catalyst.expressions.{Expression, IntegerLiteral, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.{Rule, RuleExecutor}
 
 class RuleExecutorSuite extends SparkFunSuite {
@@ -50,7 +52,7 @@ class RuleExecutorSuite extends SparkFunSuite {
       val batches = Batch("fixedPoint", FixedPoint(10), DecrementLiterals) :: Nil
     }
 
-    val message = intercept[SparkException] {
+    val message = intercept[TreeNodeException[LogicalPlan]] {
       ToFixedPoint.execute(Literal(100))
     }.getMessage
     assert(message.contains("Max iterations (10) reached for batch fixedPoint"))