[SPARK-10623][SQL]: fix the predicate pushdown construction

sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFilters.scala

@@ -33,13 +33,13 @@ import org.apache.spark.sql.sources._
 private[orc] object OrcFilters extends Logging {
   def createFilter(expr: Array[Filter]): Option[SearchArgument] = {
     expr.reduceOption(And).flatMap { conjunction =>
-      val builder = SearchArgumentFactory.newBuilder()
-      buildSearchArgument(conjunction, builder).map(_.build())
+      val builder = SearchArgumentFactory.newBuilder().startAnd()
+      buildSearchArgument(conjunction, builder).map(_.end().build())
     }
   }
 
   private def buildSearchArgument(expression: Filter, builder: Builder): Option[Builder] = {
-    def newBuilder = SearchArgumentFactory.newBuilder()
+    def newBuilder = SearchArgumentFactory.newBuilder().startAnd()
 
     def isSearchableLiteral(value: Any): Boolean = value match {
       // These are types recognized by the `SearchArgumentImpl.BuilderImpl.boxLiteral()` method.
@@ -72,8 +72,8 @@ private[orc] object OrcFilters extends Logging {
 
     expression match {
       case And(left, right) =>
-        val tryLeft = buildSearchArgument(left, newBuilder)
-        val tryRight = buildSearchArgument(right, newBuilder)
+        val tryLeft = buildSearchArgument(left, newBuilder).map(_.end)
+        val tryRight = buildSearchArgument(right, newBuilder).map(_.end)
 
         val conjunction = for {
           _ <- tryLeft
@@ -90,15 +90,15 @@ private[orc] object OrcFilters extends Logging {
 
       case Or(left, right) =>
         for {
-          _ <- buildSearchArgument(left, newBuilder)
-          _ <- buildSearchArgument(right, newBuilder)
+          _ <- buildSearchArgument(left, newBuilder).map(_.end)
+          _ <- buildSearchArgument(right, newBuilder).map(_.end)
           lhs <- buildSearchArgument(left, builder.startOr())
           rhs <- buildSearchArgument(right, lhs)
         } yield rhs.end()
 
       case Not(child) =>
         for {
-          _ <- buildSearchArgument(child, newBuilder)
+          _ <- buildSearchArgument(child, newBuilder).map(_.end())
           negate <- buildSearchArgument(child, builder.startNot())
         } yield negate.end()
 

sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcQuerySuite.scala

@@ -101,7 +101,7 @@ class OrcQuerySuite extends QueryTest with BeforeAndAfterAll with OrcTest {
     }
   }
 
-  test("Simple selection form ORC table") {
+  def simpleSelection(): Unit = {
     val data = (1 to 10).map { i =>
       Person(s"name_$i", i, (0 to 1).map { m => Contact(s"contact_$m", s"phone_$m") })
     }
@@ -147,6 +147,16 @@ class OrcQuerySuite extends QueryTest with BeforeAndAfterAll with OrcTest {
     }
   }
 
+  test("Simple selection form ORC table") {
+    simpleSelection()
+  }
+
+  test("Simple selection form ORC table with PPD") {
+    withPPD {
+      simpleSelection
+    }
+  }
+
   test("save and load case class RDD with `None`s as orc") {
     val data = (
       None: Option[Int],

sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcTest.scala

@@ -42,6 +42,12 @@ private[sql] trait OrcTest extends SQLTestUtils with TestHiveSingleton {
     }
   }
 
+  protected def withPPD(f: () => Unit) {
+    sqlContext.setConf("spark.sql.orc.filterPushdown", "true")
+    f()
+    sqlContext.setConf("spark.sql.orc.filterPushdown", "false")
+  }
+
   /**
    * Writes `data` to a Orc file and reads it back as a [[DataFrame]],
    * which is then passed to `f`. The Orc file will be deleted after `f` returns.