Fix pattern with casts and add more test cases.

Author: nongli

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala

@@ -203,9 +203,20 @@ object ExtractNonNullableAttributes extends Logging with PredicateHelper {
           result.add(b)
         }
       }
+      case BinaryComparison(Cast(a: Attribute, _), Cast(b: Attribute, _)) => {
+        if (!e.isInstanceOf[EqualNullSafe]) {
+          result.add(a)
+          result.add(b)
+        }
+      }
       case BinaryComparison(a: Attribute, _) => if (!e.isInstanceOf[EqualNullSafe]) result.add(a)
       case BinaryComparison(_, a: Attribute) => if (!e.isInstanceOf[EqualNullSafe]) result.add(a)
+      case BinaryComparison(Cast(a: Attribute, _), _) =>
+        if (!e.isInstanceOf[EqualNullSafe]) result.add(a)
+      case BinaryComparison(_, Cast(a: Attribute, _)) =>
+        if (!e.isInstanceOf[EqualNullSafe]) result.add(a)
       case Not(child) => extract(child)
+      case _ =>
     }
     predicates.foreach { extract(_) }
     result.toSet

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinFilterSuite.scala

@@ -24,6 +24,7 @@ import org.apache.spark.sql.catalyst.plans.{Inner, PlanTest}
 import org.apache.spark.sql.catalyst.rules._
 import org.apache.spark.sql.catalyst.dsl.plans._
 import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.types.DoubleType
 
 class JoinFilterSuite extends PlanTest {
 
@@ -64,6 +65,37 @@ class JoinFilterSuite extends PlanTest {
     comparePlans(optimized, correctAnswer)
   }
 
+  test("joins infer is NOT NULL one key") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+
+    val originalQuery = x.join(y).
+      where("x.b".attr + 1 === "y.b".attr)
+
+    val optimized = Optimize.execute(originalQuery.analyze)
+
+    val correctAnswer = x.join(
+        Filter(IsNotNull("y.b".attr), y), Inner, Some("x.b".attr + 1 === "y.b".attr)).analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("joins infer is NOT NULL for cast") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+
+    val originalQuery = x.join(y).
+      where(Cast("x.b".attr, DoubleType) === "y.b".attr)
+
+    val optimized = Optimize.execute(originalQuery.analyze)
+
+    val correctAnswer =
+      Filter(IsNotNull("x.b".attr), x).join(
+        Filter(IsNotNull("y.b".attr), y), Inner,
+            Some(Cast("x.b".attr, DoubleType) === Cast("y.b".attr, DoubleType))).analyze
+    comparePlans(optimized, correctAnswer)
+  }
+
   test("joins infer is NOT NULL on join keys") {
     val x = testRelation.subquery('x)
     val y = testRelation.subquery('y)