[SPARK-13891] [SQL] [TEST] Issue Exceptions when Hitting the Max Iteration Limit in Optimizer and Analyzer

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

@@ -21,9 +21,10 @@ import scala.collection.JavaConverters._
 
 import com.google.common.util.concurrent.AtomicLongMap
 
-import org.apache.spark.Logging
+import org.apache.spark.{Logging, SparkException}
 import org.apache.spark.sql.catalyst.trees.TreeNode
 import org.apache.spark.sql.catalyst.util.sideBySide
+import org.apache.spark.util.Utils
 
 object RuleExecutor {
   protected val timeMap = AtomicLongMap.create[String]()
@@ -46,15 +47,23 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
   /**
    * An execution strategy for rules that indicates the maximum number of executions. If the
    * execution reaches fix point (i.e. converge) before maxIterations, it will stop.
+   * If throws is equal to true, it will issue an exception AnalysisExceptions
    */
-  abstract class Strategy { def maxIterations: Int }
+  abstract class Strategy {
+    def maxIterations: Int
+    def throws: Boolean
+  }
 
   /** A strategy that only runs once. */
-  case object Once extends Strategy { val maxIterations = 1 }
+  case object Once extends Strategy {
+    val maxIterations = 1
+    val throws = false
+  }
 
   /** A strategy that runs until fix point or maxIterations times, whichever comes first. */
-  case class FixedPoint(maxIterations: Int) extends Strategy
-
+  case class FixedPoint(maxIterations: Int) extends Strategy {
+    override val throws: Boolean = if (Utils.isTesting) true else false
+  }
   /** A batch of rules. */
   protected case class Batch(name: String, strategy: Strategy, rules: Rule[TreeType]*)
 
@@ -98,7 +107,8 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
         if (iteration > batch.strategy.maxIterations) {
           // Only log if this is a rule that is supposed to run more than once.
           if (iteration != 2) {
-            logInfo(s"Max iterations (${iteration - 1}) reached for batch ${batch.name}")
+            val msg = s"Max iterations (${iteration - 1}) reached for batch ${batch.name}"
+            if (batch.strategy.throws) throw new SparkException(msg) else logTrace(msg)
           }
           continue = false
         }

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

@@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.rules.RuleExecutor
 class AggregateOptimizeSuite extends PlanTest {
 
   object Optimize extends RuleExecutor[LogicalPlan] {
+    System.setProperty("spark.testing", "true")
     val batches = Batch("Aggregate", FixedPoint(100),
       RemoveLiteralFromGroupExpressions) :: Nil
   }

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

@@ -29,6 +29,7 @@ import org.apache.spark.sql.catalyst.rules._
 class BooleanSimplificationSuite extends PlanTest with PredicateHelper {
 
   object Optimize extends RuleExecutor[LogicalPlan] {
+    System.setProperty("spark.testing", "true")
     val batches =
       Batch("AnalysisNodes", Once,
         EliminateSubqueryAliases) ::

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

@@ -33,6 +33,7 @@ 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,6 +26,7 @@ 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,6 +25,7 @@ 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,9 @@ import org.apache.spark.sql.types.IntegerType
 class SimplifyConditionalSuite extends PlanTest with PredicateHelper {
 
   object Optimize extends RuleExecutor[LogicalPlan] {
-    val batches = Batch("SimplifyConditionals", FixedPoint(50), SimplifyConditionals) :: Nil
+    System.setProperty("spark.testing", "true")
+    val batches =
+      Batch("SimplifyConditionals", FixedPoint(50), SimplifyConditionals) :: Nil
   }
 
   protected def assertEquivalent(e1: Expression, e2: Expression): Unit = {

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

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.trees
 
-import org.apache.spark.SparkFunSuite
+import org.apache.spark.{SparkException, SparkFunSuite}
 import org.apache.spark.sql.catalyst.expressions.{Expression, IntegerLiteral, Literal}
 import org.apache.spark.sql.catalyst.rules.{Rule, RuleExecutor}
 
@@ -46,9 +46,13 @@ class RuleExecutorSuite extends SparkFunSuite {
 
   test("to maxIterations") {
     object ToFixedPoint extends RuleExecutor[Expression] {
+      System.setProperty("spark.testing", "true")
       val batches = Batch("fixedPoint", FixedPoint(10), DecrementLiterals) :: Nil
     }
 
-    assert(ToFixedPoint.execute(Literal(100)) === Literal(90))
+    val message = intercept[SparkException] {
+      ToFixedPoint.execute(Literal(100))
+    }.getMessage
+    assert(message.contains("Max iterations (10) reached for batch fixedPoint"))
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala

@@ -84,6 +84,7 @@ private[sql] trait SQLTestUtils
 
   protected override def beforeAll(): Unit = {
     super.beforeAll()
+    System.setProperty("spark.testing", "true")
     if (loadTestDataBeforeTests) {
       loadTestData()
     }