[SPARK-22548][SQL] Incorrect nested AND expression pushed down to JDBC data source

Author: jliwork

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala

@@ -497,7 +497,10 @@ object DataSourceStrategy {
         Some(sources.IsNotNull(a.name))
 
       case expressions.And(left, right) =>
-        (translateFilter(left) ++ translateFilter(right)).reduceOption(sources.And)
+        for {
+          leftFilter <- translateFilter(left)
+          rightFilter <- translateFilter(right)
+        } yield sources.And(leftFilter, rightFilter)
 
       case expressions.Or(left, right) =>
         for {

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

@@ -296,8 +296,33 @@ class JDBCSuite extends SparkFunSuite
     // The older versions of spark have this kind of bugs in parquet data source.
     val df1 = sql("SELECT * FROM foobar WHERE NOT (THEID != 2 AND NAME != 'mary')")
     val df2 = sql("SELECT * FROM foobar WHERE NOT (THEID != 2) OR NOT (NAME != 'mary')")
+    val df3 = sql("SELECT * FROM foobar WHERE (THEID > 0 AND NAME = 'mary') OR (NAME = 'fred')")
+    val df4 = sql("SELECT * FROM foobar " +
+      "WHERE (THEID > 0 AND TRIM(NAME) = 'mary') OR (NAME = 'fred')")
+    val df5 = sql("SELECT * FROM foobar " +
+      "WHERE THEID > 0 AND TRIM(NAME) = 'mary' AND LENGTH(NAME) > 3")
+    val df6 = sql("SELECT * FROM foobar " +
+      "WHERE THEID < 0 OR NAME = 'mary' OR NAME = 'fred'")
+    val df7 = sql("SELECT * FROM foobar " +
+      "WHERE THEID < 0 OR TRIM(NAME) = 'mary' OR NAME = 'fred'")
+    val df8 = sql("SELECT * FROM foobar " +
+      "WHERE NOT((THEID < 0 OR NAME != 'mary') AND (THEID != 1 OR NAME != 'fred'))")
+    val df9 = sql("SELECT * FROM foobar " +
+      "WHERE NOT((THEID < 0 OR NAME != 'mary') AND (THEID != 1 OR TRIM(NAME) != 'fred'))")
+    val df10 = sql("SELECT * FROM foobar " +
+      "WHERE (NOT(THEID < 0 OR TRIM(NAME) != 'mary')) OR (THEID = 1 AND NAME = 'fred')")
+
     assert(df1.collect.toSet === Set(Row("mary", 2)))
     assert(df2.collect.toSet === Set(Row("mary", 2)))
+    assert(df3.collect.toSet === Set(Row("fred", 1), Row("mary", 2)))
+    assert(df4.collect.toSet === Set(Row("fred", 1), Row("mary", 2)))
+    assert(df5.collect.toSet === Set(Row("mary", 2)))
+    assert(df6.collect.toSet === Set(Row("fred", 1), Row("mary", 2)))
+    assert(df7.collect.toSet === Set(Row("fred", 1), Row("mary", 2)))
+    assert(df8.collect.toSet === Set(Row("fred", 1), Row("mary", 2)))
+    assert(df9.collect.toSet === Set(Row("fred", 1), Row("mary", 2)))
+    assert(df10.collect.toSet === Set(Row("fred", 1), Row("mary", 2)))
+
 
     def checkNotPushdown(df: DataFrame): DataFrame = {
       val parentPlan = df.queryExecution.executedPlan