address comments

dongjoon-hyun · dongjoon-hyun · commit fe9d98bf990a · 2017-06-10T12:34:14.000-07:00
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
@@ -110,7 +110,6 @@ abstract class Optimizer(sessionCatalog: SessionCatalog, conf: SQLConf)
       EliminateSerialization,
       RemoveRedundantAliases,
       RemoveRedundantProject,
-      RemoveInvalidRange,
       SimplifyCreateStructOps,
       SimplifyCreateArrayOps,
       SimplifyCreateMapOps,
@@ -271,16 +270,6 @@ object RemoveRedundantProject extends Rule[LogicalPlan] {
   }
 }
 
-/**
- * Replace invalid `range` with emtpy range.
- */
-object RemoveInvalidRange extends Rule[LogicalPlan] {
-  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
-    case p @ Range(start, end, step, a, b) if (start == end) || (start < end ^ 0 < step) =>
-      LocalRelation(p.output, data = Seq.empty)
-  }
-}
-
 /**
  * Pushes down [[LocalLimit]] beneath UNION ALL and beneath the streamed inputs of outer joins.
  */
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
@@ -482,11 +482,17 @@ case class Sort(
 
 /** Factory for constructing new `Range` nodes. */
 object Range {
-  def apply(start: Long, end: Long, step: Long, numSlices: Option[Int]): Range = {
+  def apply(start: Long, end: Long, step: Long, numSlices: Option[Int])
+      : LeafNode with MultiInstanceRelation = {
     val output = StructType(StructField("id", LongType, nullable = false) :: Nil).toAttributes
-    new Range(start, end, step, numSlices, output)
+    if (start == end || (start < end ^ 0 < step)) {
+      LocalRelation(output)
+    } else {
+      new Range(start, end, step, numSlices, output)
+    }
   }
-  def apply(start: Long, end: Long, step: Long, numSlices: Int): Range = {
+  def apply(start: Long, end: Long, step: Long, numSlices: Int)
+      : LeafNode with MultiInstanceRelation = {
     Range(start, end, step, Some(numSlices))
   }
 }
@@ -500,6 +506,8 @@ case class Range(
   extends LeafNode with MultiInstanceRelation {
 
   require(step != 0, s"step ($step) cannot be 0")
+  require(start != end, s"start ($step) cannot be equal to end ($end)")
+  require(start < end ^ step < 0, s"the sign of step ($step) is invalid for range ($start, $end)")
 
   val numElements: BigInt = {
     val safeStart = BigInt(start)
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RemoveInvalidRangeSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RemoveInvalidRangeSuite.scala
