Adding disjunctive filter predicates

Author: AndreSchumacher

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala

@@ -45,7 +45,7 @@ case class BoundReference(ordinal: Int, baseReference: Attribute)
 
   override def toString = s"$baseReference:$ordinal"
 
-  override def eval(input: Row): Any = if (input != null) input(ordinal) else null
+  override def eval(input: Row): Any = input(ordinal)
 }
 
 /**

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetFilters.scala

@@ -114,6 +114,18 @@ object ParquetFilters {
     // https://github.com/Parquet/parquet-mr/issues/371
     // has been resolved
     val filters: Seq[UnboundRecordFilter] = filterExpressions.collect {
+      case Or(left: Expression, right: Expression)
+          if createFilter(Seq(left)) != null && createFilter(Seq(right)) != null => {
+        // Note: if either side of this Or-predicate is empty then this means
+        // it contains a more complex comparison than between attribute and literal
+        // (e.g., it contained a CAST). The only safe thing to do is then to disregard
+        // this disjunction, which could be contained in a conjunction. If it stands
+        // alone then it is also safe to drop it, since a Null return value of this
+        // function is interpreted as having no filters at all.
+        val leftFilter = createFilter(Seq(left))
+        val rightFilter = createFilter(Seq(right))
+        OrRecordFilter.or(leftFilter, rightFilter)
+      }
       case Equals(left: Literal, right: NamedExpression) if !right.nullable =>
         createEqualityFilter(right.name, left)
       case Equals(left: NamedExpression, right: Literal) if !left.nullable =>
@@ -135,7 +147,6 @@ object ParquetFilters {
       case GreaterThanOrEqual(left: NamedExpression, right: Literal) if !left.nullable =>
         createGreaterThanOrEqualFilter(left.name, right)
     }
-    // TODO: How about disjunctions? (Or-ed)
     if (filters.length > 0) filters.reduce(AndRecordFilter.and) else null
   }
 

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetRelation.scala

@@ -146,16 +146,6 @@ private[sql] object ParquetRelation {
     new ParquetRelation(path.toString)
   }
 
-  def checkPredicatePushdownPossible(filters: Seq[Expression]): Boolean = {
-    def checkFeasible(left: Expression, right: Expression) =
-      left.isInstanceOf[Literal] || right.isInstanceOf[Literal]
-    filters.forall {
-      case Equals(left, right) => checkFeasible(left, right)
-      case LessThan(left, right) => checkFeasible(left, right)
-      case _ => false
-    }
-  }
-
   private def checkPath(pathStr: String, allowExisting: Boolean, conf: Configuration): Path = {
     if (pathStr == null) {
       throw new IllegalArgumentException("Unable to create ParquetRelation: path is null")

sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetQuerySuite.scala

@@ -266,6 +266,14 @@ class ParquetQuerySuite extends QueryTest with FunSuite with BeforeAndAfterAll {
       val result2 = sql(s"SELECT * FROM testfiltersource WHERE $myval > 150 AND $myval <= 200").collect()
       assert(result2.size === 50)
     }
+    for(myval <- Seq("myint", "mylong")) {
+      val result3 = sql(s"SELECT * FROM testfiltersource WHERE $myval > 190 OR $myval < 10").collect()
+      assert(result3.size === 20)
+    }
+    for(myval <- Seq("mydouble", "myfloat")) {
+      val result3 = sql(s"SELECT * FROM testfiltersource WHERE $myval > 190.5 OR $myval < 10").collect()
+      assert(result3.size === 20)
+    }
     val booleanResult = sql(s"SELECT * FROM testfiltersource WHERE myboolean = true AND myint < 40").collect()
     assert(booleanResult.size === 10)
     val stringResult = sql("SELECT * FROM testfiltersource WHERE mystring = \"100\"").collect()