[SPARK-21044][SPARK-21041][SQL] Add RemoveInvalidRange optimizer

dongjoon-hyun · dongjoon-hyun · commit 6ab9b6fb559a · 2017-06-09T18:53:24.000-07:00
**BEFORE**
```
scala&gt; spark.range(0,0,1).explain
== Physical Plan ==
*Range (0, 0, step=1, splits=8)
```

**AFTER**
```
scala&gt; spark.range(0,0,1).explain
== Physical Plan ==
LocalTableScan &lt;empty&gt;, [id#0L]
```
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,6 +110,7 @@ abstract class Optimizer(sessionCatalog: SessionCatalog, conf: SQLConf)
       EliminateSerialization,
       RemoveRedundantAliases,
       RemoveRedundantProject,
+      RemoveInvalidRange,
       SimplifyCreateStructOps,
       SimplifyCreateArrayOps,
       SimplifyCreateMapOps,
@@ -270,6 +271,16 @@ 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/test/scala/org/apache/spark/sql/catalyst/optimizer/RemoveInvalidRangeSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RemoveInvalidRangeSuite.scala
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.plans._
+import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan, Range}
+import org.apache.spark.sql.catalyst.rules.RuleExecutor
+
+class RemoveInvalidRangeSuite extends PlanTest {
+  object Optimize extends RuleExecutor[LogicalPlan] {
+    val batches =
+      Batch("RemoveInvalidRange", Once,
+        RemoveInvalidRange) :: Nil
+  }
+
+  test("preserve valid ranges") {
+    Seq(Range(0, 10, 1, 1), Range(10, 0, -1, 1)).foreach { query =>
+      val optimized = Optimize.execute(query.analyze)
+      val correctAnswer = query
+
+      comparePlans(optimized, correctAnswer)
+    }
+  }
+
+  test("remove ranges with invalid combination of start/end/step") {
+    Seq(Range(0, 0, 1, 1), Range(0, 0, -1, 1), Range(1, 10, -1, 1), Range(10, 1, 1, 1)).foreach {
+      query =>
+        val optimized = Optimize.execute(query.analyze)
+        val correctAnswer = LocalRelation(query.output, data = Seq.empty)
+        comparePlans(optimized, correctAnswer)
+    }
+  }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala
@@ -191,6 +191,17 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext with Eventuall
       checkAnswer(sql("SELECT * FROM range(3)"), Row(0) :: Row(1) :: Row(2) :: Nil)
     }
   }
+
+  test("SPARK-21041 SparkSession.range()'s behavior is inconsistent with SparkContext.range()") {
+    val start = java.lang.Long.MAX_VALUE - 3
+    val end = java.lang.Long.MIN_VALUE + 2
+    Seq("false", "true").foreach { value =>
+      withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> value) {
+        assert(spark.sparkContext.range(start, end, 1).collect.length == 0)
+        assert(spark.range(start, end, 1).collect.length == 0)
+      }
+    }
+  }
 }
 
 object DataFrameRangeSuite {

Original file line number	Diff line number	Diff line change
`@@ -191,6 +191,17 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext with Eventuall`
`191`	`191`	`checkAnswer(sql("SELECT * FROM range(3)"), Row(0) :: Row(1) :: Row(2) :: Nil)`
`192`	`192`	`}`
`193`	`193`	`}`
	`194`	`+`
	`195`	`+ test("SPARK-21041 SparkSession.range()'s behavior is inconsistent with SparkContext.range()") {`
	`196`	`+ val start = java.lang.Long.MAX_VALUE - 3`
	`197`	`+ val end = java.lang.Long.MIN_VALUE + 2`
	`198`	`+ Seq("false", "true").foreach { value =>`
	`199`	`+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> value) {`
	`200`	`+ assert(spark.sparkContext.range(start, end, 1).collect.length == 0)`
	`201`	`+ assert(spark.range(start, end, 1).collect.length == 0)`
	`202`	`+ }`
	`203`	`+ }`
	`204`	`+ }`
`194`	`205`	`}`
`195`	`206`
`196`	`207`	`object DataFrameRangeSuite {`