improve OrcFilters in Hive project as well

Author: gengliangwang

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

@@ -155,12 +155,23 @@ private[orc] object OrcFilters extends Logging {
         }
 
       case Or(left, right) =>
+        // The Or predicate is convertible when both of its children can be pushed down.
+        // That is to say, if one/both of the children can be partially pushed down, the Or
+        // predicate can be partially pushed down as well.
+        //
+        // Here is an example used to explain the reason.
+        // Let's say we have
+        // (a1 AND a2) OR (b1 AND b2),
+        // a1 and b1 is convertible, while a2 and b2 is not.
+        // The predicate can be converted as
+        // (a1 OR b1) AND (a1 OR b2) AND (a2 OR b1) AND (a2 OR b2)
+        // As per the logical in And predicate, we can push down (a1 OR b1).
         for {
-          _ <- createBuilder(dataTypeMap, left, newBuilder, canPartialPushDownConjuncts = false)
-          _ <- createBuilder(dataTypeMap, right, newBuilder, canPartialPushDownConjuncts = false)
+          _ <- createBuilder(dataTypeMap, left, newBuilder, canPartialPushDownConjuncts)
+          _ <- createBuilder(dataTypeMap, right, newBuilder, canPartialPushDownConjuncts)
           lhs <- createBuilder(dataTypeMap, left,
-            builder.startOr(), canPartialPushDownConjuncts = false)
-          rhs <- createBuilder(dataTypeMap, right, lhs, canPartialPushDownConjuncts = false)
+            builder.startOr(), canPartialPushDownConjuncts)
+          rhs <- createBuilder(dataTypeMap, right, lhs, canPartialPushDownConjuncts)
         } yield rhs.end()
 
       case Not(child) =>

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

@@ -399,17 +399,6 @@ class HiveOrcFilterSuite extends OrcTest with TestHiveSingleton {
       )).get.toString
     }
 
-    // Can not remove unsupported `StringContains` predicate since it is under `Or` operator.
-    assert(OrcFilters.createFilter(schema, Array(
-      Or(
-        LessThan("a", 10),
-        And(
-          StringContains("b", "prefix"),
-          GreaterThan("a", 1)
-        )
-      )
-    )).isEmpty)
-
     // Safely remove unsupported `StringContains` predicate and push down `LessThan`
     assertResultWithDiffHiveVersion(
       """leaf-0 = (LESS_THAN a 10)
@@ -442,4 +431,53 @@ class HiveOrcFilterSuite extends OrcTest with TestHiveSingleton {
       )).get.toString
     }
   }
+
+  test("SPARK-27699 Converting disjunctions into ORC SearchArguments") {
+    import org.apache.spark.sql.sources._
+    // The `LessThan` should be converted while the `StringContains` shouldn't
+    val schema = new StructType(
+      Array(
+        StructField("a", IntegerType, nullable = true),
+        StructField("b", StringType, nullable = true)))
+
+    // The predicate `StringContains` predicate is not able to be pushed down.
+    assertResult("leaf-0 = (LESS_THAN_EQUALS a 10)\nleaf-1 = (LESS_THAN a 1)\n" +
+      "expr = (or (not leaf-0) leaf-1)") {
+      OrcFilters.createFilter(schema, Array(
+        Or(
+          GreaterThan("a", 10),
+          And(
+            StringContains("b", "prefix"),
+            LessThan("a", 1)
+          )
+        )
+      )).get.toString
+    }
+
+    assertResult("leaf-0 = (LESS_THAN_EQUALS a 10)\nleaf-1 = (LESS_THAN a 1)\n" +
+      "expr = (or (not leaf-0) leaf-1)") {
+      OrcFilters.createFilter(schema, Array(
+        Or(
+          And(
+            GreaterThan("a", 10),
+            StringContains("b", "foobar")
+          ),
+          And(
+            StringContains("b", "prefix"),
+            LessThan("a", 1)
+          )
+        )
+      )).get.toString
+    }
+
+    assert(OrcFilters.createFilter(schema, Array(
+      Or(
+        StringContains("b", "foobar"),
+        And(
+          StringContains("b", "prefix"),
+          LessThan("a", 1)
+        )
+      )
+    )).isEmpty)
+  }
 }