From b5585493966ea94c9926b878a5d87ed56f62e6e3 Mon Sep 17 00:00:00 2001
From: Wenchen Fan <cloud0fan@outlook.com>
Date: Sat, 13 Jun 2015 10:31:23 +0800
Subject: [PATCH] address comments

---
 .../sql/catalyst/optimizer/Optimizer.scala      | 17 +++++++++++------
 ...hDownSuit.scala => LimitPushDownSuite.scala} |  2 +-
 2 files changed, 12 insertions(+), 7 deletions(-)
 rename sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/{LimitPushDownSuit.scala => LimitPushDownSuite.scala} (98%)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index 92f79dab7be58..57d25069ef548 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -39,20 +39,23 @@ object DefaultOptimizer extends Optimizer {
     Batch("Distinct", FixedPoint(100),
       ReplaceDistinctWithAggregate) ::
     Batch("Operator Optimizations", FixedPoint(100),
+      // Operator push down
       UnionPushDown,
-      CombineFilters,
+      LimitPushDown,
+      PushPredicateThroughJoin,
       PushPredicateThroughProject,
       PushPredicateThroughGenerate,
       ColumnPruning,
-      LimitPushDown,
+      // Operator combine
       ProjectCollapsing,
+      CombineFilters,
       CombineLimits,
+      // Constant folding
       NullPropagation,
       OptimizeIn,
       ConstantFolding,
       LikeSimplification,
       BooleanSimplification,
-      PushPredicateThroughJoin,
       RemovePositive,
       SimplifyFilters,
       SimplifyCasts,
@@ -111,12 +114,14 @@ object UnionPushDown extends Rule[LogicalPlan] {
 
 object LimitPushDown extends Rule[LogicalPlan] {
   def apply(plan: LogicalPlan): LogicalPlan = plan transform {
-    // Push down limit when the child is project on limit
+    // Push down limit when the child is project on limit.
     case Limit(expr, Project(projectList, l: Limit)) =>
       Project(projectList, Limit(expr, l))
 
-    // Push down limit when the child is project on sort
-    case Limit(expr, Project(projectList, s: Sort)) =>
+    // Push down limit when the child is project on sort,
+    // and we cannot push down this project through sort.
+    case Limit(expr, p @ Project(projectList, s: Sort))
+      if !s.references.subsetOf(p.outputSet) =>
       Project(projectList, Limit(expr, s))
   }
 }
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushDownSuit.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushDownSuite.scala
similarity index 98%
rename from sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushDownSuit.scala
rename to sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushDownSuite.scala
index 5d774efc63c62..c8168b32eef94 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushDownSuit.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushDownSuite.scala
@@ -24,7 +24,7 @@ import org.apache.spark.sql.catalyst.rules.RuleExecutor
 import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.dsl.plans._
 
-class LimitPushDownSuit extends PlanTest {
+class LimitPushDownSuite extends PlanTest {
 
   object Optimize extends RuleExecutor[LogicalPlan] {
     val batches =