[SQL] Better logging when applying rules.

Author: Michael Armbrust <michael@databricks.com>

Closes #616 from marmbrus/ruleLogging and squashes the following commits:

39c09fe [Michael Armbrust] Fix off by one error.
5af3537 [Michael Armbrust] Better logging when applying rules.

1 files changed, 24 insertions, 4 deletions

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala
index 9db96f89dd..e32adb76fe 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala
@@ -50,18 +50,18 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
     var curPlan = plan
 
     batches.foreach { batch =>
+      val batchStartPlan = curPlan
       var iteration = 1
       var lastPlan = curPlan
-      curPlan = batch.rules.foldLeft(curPlan) { case (plan, rule) => rule(plan) }
+      var continue = true
 
       // Run until fix point (or the max number of iterations as specified in the strategy.
-      while (iteration < batch.strategy.maxIterations && !curPlan.fastEquals(lastPlan)) {
-        lastPlan = curPlan
+      while (continue) {
         curPlan = batch.rules.foldLeft(curPlan) {
           case (plan, rule) =>
             val result = rule(plan)
             if (!result.fastEquals(plan)) {
-              logger.debug(
+              logger.trace(
                 s"""
                   |=== Applying Rule ${rule.ruleName} ===
                   |${sideBySide(plan.treeString, result.treeString).mkString("\n")}
@@ -71,6 +71,26 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
             result
         }
         iteration += 1
+        if (iteration > batch.strategy.maxIterations) {
+          logger.info(s"Max iterations ($iteration) reached for batch ${batch.name}")
+          continue = false
+        }
+
+        if (curPlan.fastEquals(lastPlan)) {
+          logger.trace(s"Fixed point reached for batch ${batch.name} after $iteration iterations.")
+          continue = false
+        }
+        lastPlan = curPlan
+      }
+
+      if (!batchStartPlan.fastEquals(curPlan)) {
+        logger.debug(
+          s"""
+          |=== Result of Batch ${batch.name} ===
+          |${sideBySide(plan.treeString, curPlan.treeString).mkString("\n")}
+        """.stripMargin)
+      } else {
+        logger.trace(s"Batch ${batch.name} has no effect.")
       }
     }